Data Import does not consume imported ids

### [REQUIRED] Environment info

**firebase-tools:** 14.25.0

**Platform:** macOS

### [REQUIRED] Test case

```
#!/usr/bin/env python3
import logging
import os
import shutil
import subprocess
import time

import psutil
import requests

import portpicker

from gcloud.rest import datastore

# Import the FirestoreEmulator class from the local copy
from datastore_emulator import FirestoreEmulator

# Set up logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
log = logging.getLogger(__name__)

def create_person_entity(project_id: str, name: str, age: int, email: str) -> dict:
"""Create a Person entity with auto-assigned ID.

Args:
project_id: The Firestore project ID
name: The person's name
age: The person's age
email: The person's email

Returns:
A Person entity dict with no name field in the key (uses auto-assigned ID)
"""
return {
'key': {
'partitionId': {'projectId': project_id},
'path': [{'kind': 'Person'}] # No 'name' field = auto-assigned ID
},
'properties': {
'name': {'stringValue': name},
'age': {'integerValue': age},
'email': {'stringValue': email}
}
}

def create_book_entity(project_id: str, title: str, author: str, year: int) -> dict:
"""Create a Book entity with auto-assigned ID.

Args:
project_id: The Firestore project ID
title: The book's title
author: The book's author
year: The publication year

Returns:
A Book entity dict with no name field in the key (uses auto-assigned ID)
"""
return {
'key': {
'partitionId': {'projectId': project_id},
'path': [{'kind': 'Book'}] # No 'name' field = auto-assigned ID
},
'properties': {
'title': {'stringValue': title},
'author': {'stringValue': author},
'year': {'integerValue': year}
}
}

def add_test_data(project_id: str):
"""Add some test entities to the emulator."""
log.info("Adding test data to the emulator...")

# Create a datastore client
client = datastore.Datastore(project=project_id)

# Create some test entities using helper methods
entities = []

# Create a few Person entities
for i in range(1, 4):
entity = create_person_entity(
project_id=project_id,
name=f'Person {i}',
age=20 + i,
email=f'person{i}@example.com'
)
entities.append(entity)

# Create a few Book entities
for i in range(1, 3):
entity = create_book_entity(
project_id=project_id,
title=f'Book Title {i}',
author=f'Author {i}',
year=2020 + i
)
entities.append(entity)

# Commit the entities
mutations = [{'upsert': entity} for entity in entities]

# Use the REST API directly
import requests
emulator_host = os.environ['DATASTORE_EMULATOR_HOST']
url = f'http://{emulator_host}/v1/projects/{project_id}:commit'

payload = {
'mode': 'NON_TRANSACTIONAL',
'mutations': mutations
}

response = requests.post(url, json=payload)
if response.status_code == 200:
log.info(f"Successfully added {len(entities)} test entities")
else:
log.error(f"Failed to add entities: {response.text}")
raise Exception(f"Failed to add entities: {response.text}")

def add_more_data(project_id: str):
entity = create_person_entity(
project_id=project_id,
name=f'Juan',
age=30,
email=f'person@example.com'
)
# Commit the entities
mutations = [{'upsert': entity}]

# Use the REST API directly
import requests
emulator_host = os.environ['DATASTORE_EMULATOR_HOST']
url = f'http://{emulator_host}/v1/projects/{project_id}:commit'

payload = {
'mode': 'NON_TRANSACTIONAL',
'mutations': mutations
}

response = requests.post(url, json=payload)
if response.status_code == 200:
log.info(f"Successfully added 1 test entities")
else:
log.error(f"Failed to add entities: {response.text}")
raise Exception(f"Failed to add entities: {response.text}")

def query_data(project_id: str) -> list:
"""Query all entities from the emulator."""
log.info("Querying data from the emulator...")

import requests
emulator_host = os.environ['DATASTORE_EMULATOR_HOST']

entities = []

# Query Person entities
for kind in ['Person', 'Book']:
url = f'http://{emulator_host}/v1/projects/{project_id}:runQuery'
payload = {
'query': {
'kind': [{'name': kind}]
}
}

response = requests.post(url, json=payload)
if response.status_code == 200:
result = response.json()
batch = result.get('batch', {})
entity_results = batch.get('entityResults', [])
entities.extend(entity_results)
log.info(f"Found {len(entity_results)} {kind} entities")
else:
log.error(f"Failed to query {kind}: {response.text}")

return entities

def export_datastore_emulator_data(export_path: str, emulator_host : str, datastore_project_id: str, session: requests.Session) -> None:
"""Export emulator's data to a file.
https://cloud.google.com/datastore/docs/emulator#export_entities_in_the_emulator
"""
os.makedirs(export_path, exist_ok=False)
url = f'http://{emulator_host}/emulator/v1/projects/{datastore_project_id}:export'
data = {
'database': f'projects/{datastore_project_id}/databases/',
'export_directory': export_path,
}
logging.info('Trying to export Firestore emulator data with url: %s and data: %s', url, data)
response = session.post(url, timeout=5, json=data)
if response.status_code == 200:
logging.info('Firestore emulator data exported successfully.')
return
else:
log.error(response.text)
raise IOError('Failed to export data: %s' % response.text)

def reset_inmemory_data(session, emulator_host) -> None:
response = session.post(f'http://{emulator_host}/reset', timeout=5)
if response.status_code == 200:
return
else:
log.error(response.text)
raise IOError('Failed to reset in-memory data: %s' % response.text)

def import_datastore_emulator_data(emulator_host: str, datastore_project_id: str, import_path:str, session: requests.Session) -> None:
url = f'http://{emulator_host}/emulator/v1/projects/{datastore_project_id}:import'
# Find the file with extension .overall_export_metadata in the import path
overall_export_metadata = None
for root, _, files in os.walk(import_path):
if overall_export_metadata:
break
for file in files:
if file.endswith('.overall_export_metadata'):
overall_export_metadata = os.path.join(root, file)
break
data = {
'database': f'projects/{datastore_project_id}/databases/',
'export_directory': overall_export_metadata,
}
response = session.post(url, timeout=5, json=data)
if response.status_code == 200:
logging.info('Firestore emulator data imported successfully.')
# Allocate IDs to prevent reuse of imported entity IDs
return
else:
log.error(response.text)
raise IOError('Failed to import data: %s' % response.text)

def main():
# Directory for exports
export_dir = '/tmp/firestore-export-demo'

# Clean up any existing export directory
if os.path.exists(export_dir):
log.info(f"Cleaning up existing export directory: {export_dir}")
shutil.rmtree(export_dir)

# Start emulator
emulator_port = portpicker.PickUnusedPort()
emulator_host = f'127.0.0.1:{emulator_port}'
start_command = (f'gcloud emulators firestore start '
f'--database-mode=datastore-mode '
f'--host-port={emulator_host}')
process = subprocess.Popen(start_command.split(' '),
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL)
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
pass
project_id = os.getenv('DATASTORE_PROJECT_ID', 'testbed-test')
session = requests.Session()

os.environ['DATASTORE_EMULATOR_HOST'] = emulator_host
os.environ['DATASTORE_PROJECT_ID'] = project_id
if os.environ.get('GOOGLE_CLOUD_PROJECT', None) == 'dev':
# testenv:workflows sets GOOGLE_CLOUD_PROJECT=dev, and they need DATASTORE_PROJECT_ID to be None
os.environ.pop('DATASTORE_PROJECT_ID', None)

try:
# Give emulator a moment to fully initialize
time.sleep(2)

log.info("Adding test data...")
add_test_data(project_id)

log.info("Exporting data...")
export_datastore_emulator_data(export_dir, emulator_host, project_id, session)

# List the exported files
if os.path.exists(export_dir):
log.info("Exported files:")
for root, dirs, files in os.walk(export_dir):
for file in files:
file_path = os.path.join(root, file)
log.info(f" - {file_path}")

log.info("Resetting emulator data...")
reset_inmemory_data(session, emulator_host)
entities_after_reset = query_data(project_id)
log.info(f"Total entities after reset: {len(entities_after_reset)}")

log.info("Importing data back...")
import_datastore_emulator_data(emulator_host, project_id, export_dir, session)
log.info("Data import complete")

# Give it a moment to process the import
time.sleep(2)

log.info("\n" + "=" * 60)
log.info("Verifying imported data...")
final_entities = query_data(project_id)
log.info(f"Total entities after import: {len(final_entities)}")
add_more_data(project_id)
final_entities = query_data(project_id)
log.info(f"Total entities after import: {len(final_entities)}")

finally:
# Clean up
log.info("Cleaning up...")

p = psutil.Process(process.pid)
process_group = p.children(recursive=True)
process_group.append(p)

for process in process_group:
try:
process.terminate()
except psutil.NoSuchProcess:
pass
except Exception:
log.exception(
'Process: %s could not be terminated, please terminate manually if they '
'exist.', process)
session.close()
log.info("Emulator shut down successfully")

if __name__ == '__main__':
main()
```

### [REQUIRED] Steps to reproduce

In the above tests case we have the following steps:
- Start emulator and create entities
- Create an export
- Reset the DB
- Import the previous data
At this point, we can not create entites from the same Kind.
I tested adding manually allocating the ids after the import and that fixes the issue, but I expect this to be included as part of the import.

### [REQUIRED] Expected behavior

Entities of imported Kind can be created after an import

### [REQUIRED] Actual behavior

Error adding entities after the import:
```
{"error":{"code":400,"message":"the id allocated for a new entity was already in use, please try again: app: \"testbed-test\"\npath <\n Element {\n type: \"Person\"\n id: 0x14000000000000\n }\n>\n","status":"INVALID_ARGUMENT"}}
```

Start with the Python reproduction, especially import_datastore_emulator_data and the emulator /emulator/v1/projects/{project}:import endpoint. Run the export, reset, import, and subsequent same-kind entity creation steps to observe the failure. Done means importing data also prevents newly created entities from reusing imported IDs, so add_more_data succeeds after import.

Written by the indexing model from the issue text.