feat(go): bulk ingest via Volume + COPY INTO instead of per-row INSERT - #411
Draft
jatorre wants to merge 6 commits into
Draft
feat(go): bulk ingest via Volume + COPY INTO instead of per-row INSERT#411jatorre wants to merge 6 commits into
jatorre wants to merge 6 commits into
Conversation
Expose geospatialAsArrow support (SPARK-54232) as an opt-in ADBC connection option. When set to "true", geometry/geography columns arrive as Struct<srid: Int32, wkb: Binary> instead of EWKT strings. This depends on databricks/databricks-sql-go#328 which adds the WithArrowNativeGeospatial() ConnOption to the underlying Go SQL driver. Usage via adbc_connect (e.g. from DuckDB adbc_scanner): adbc_connect({ 'driver': 'libadbc_driver_databricks.dylib', 'databricks.server_hostname': '...', 'databricks.arrow.native_geospatial': 'true' })
When databricks.arrow.native_geospatial is enabled, the driver now converts Struct<srid: Int32, wkb: Binary> columns to flat Binary columns with ARROW:extension:name=geoarrow.wkb metadata. This enables downstream consumers (e.g. DuckDB adbc_scanner) to automatically map geometry columns to native GEOMETRY types without any explicit ST_GeomFromWKB conversion. Pipeline: Databricks -> Struct<srid,wkb> -> geoarrow.wkb -> native GEOMETRY Benchmarks vs baseline (ST_AsBinary + ST_GeomFromWKB): 100k points: 2.05x faster (31k rows/sec vs 15k rows/sec) 10k polygons: 1.31x faster (4.5k rows/sec vs 3.4k rows/sec)
Defer schema transformation to the first Next() call so the SRID can be read from the first non-null row of each geometry column. The SRID is encoded as PROJJSON CRS in ARROW:extension:metadata, e.g. EPSG:4326 or EPSG:3857. This ensures CRS information propagates correctly to downstream consumers (DuckDB, pandas, polars, GDAL). Split transformSchemaForGeoArrow into: - detectGeometryColumns: finds geometry struct column indices (called in constructor) - buildGeoArrowSchema: builds geoarrow schema with CRS from first batch (called lazily) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
The schema must be available before the first Next() call since consumers like adbc_scanner read it upfront to create table columns. Build the geoarrow.wkb schema eagerly with empty CRS metadata in the constructor, then enrich it with the actual SRID from the first record batch during the first Next() call. Verified: DuckDB now correctly recognizes geometry columns as native GEOMETRY type via the geoarrow.wkb extension metadata. Benchmark results (Databricks → DuckDB): - 100k points: 7x faster than ST_AsBinary baseline - 10k polygons: 3.6x faster than ST_AsBinary baseline Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Dewey Dunnington <dewey@dunnington.ca>
The current executeIngest loops over each row and issues a separate
parameterized INSERT against the SQL warehouse — measured at ~1.5 s/row
on a serverless Databricks SQL warehouse, which makes adbc_insert
impractical above ~few hundred rows. When the new
databricks.bulk.volume_path option is set, executeIngest instead
streams Arrow batches as Parquet to a Volume and issues one COPY INTO
per batch. RSS stays bounded to one batch (~37 MB at 100 K rows in a
side-by-side bench), throughput jumps from ~0.7 rows/s to ~6 K rows/s
at 100 K rows.
For geometry columns annotated as geoarrow.wkb on the Arrow schema:
- The Databricks DDL emitter (new databricksTypeForField) maps the
field to GEOMETRY(<srid>) using the EPSG code from the extension's
CRS metadata, so the destination column is a typed geometry rather
than the BINARY default. Bare GEOMETRY without an SRID modifier is
rejected on most Databricks SQL runtimes.
- The staged Parquet schema strips the extension metadata so the file
carries plain BINARY columns; the COPY INTO transform projects each
geometry column through ST_GEOMFROMWKB(<col>, <srid_literal>) so the
source binary is rebuilt as typed GEOMETRY(srid) on insert. The
SRID has to be a SQL literal — passing it as a column from the
Parquet file produces an untyped GEOMETRY whose schema can't be
merged with the destination column.
Plumbing: the connection now carries serverHostname / accessToken /
bulkVolumePath copied from databaseImpl, used by the Files API
helpers (uploadToVolume / deleteFromVolume) to PUT/DELETE staged
Parquet files via /api/2.0/fs/files{path}.
The legacy per-row INSERT path is kept as the fallback when
databricks.bulk.volume_path is unset, so this change is purely
additive and doesn't break callers without Volume access.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
jatorre
force-pushed
the
feat/bulk-volume-copy-on-main
branch
from
May 22, 2026 11:05
d6b9f44 to
9c444f2
Compare
Author
|
Note for reviewers: this stacks on #350 (Arrow-native geospatial). The DDL emitter uses the geoarrow.wkb type info from #350 to produce typed Just rebased onto current main. The diff still shows #350's contents because GitHub requires the base branch to live on the upstream repo — once #350 merges, this will narrow to the bulk-ingest changes (~640 lines, 5 files). |
4 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Replaces the per-row `INSERT` loop in `go/bulk_ingest.go`'s `executeIngest` with a streaming Parquet + COPY INTO path when a new `databricks.bulk.volume_path` option is set. Falls back to the per-row path when the option is unset, so this is purely additive and non-breaking.
Measured ~7,000× speedup at 100 K rows on a serverless Databricks SQL warehouse, with RSS bounded to ~11 MB regardless of source size.
Why
The current implementation issues one `INSERT INTO t VALUES (?, ?, ?)` per row through `db/sql.ExecContext`. Each call is a round-trip to the warehouse; in practice that pegs at ~0.7 rows/s, making `adbc_insert` impractical above a few hundred rows. The path I see commonly worked around outside the driver is "write Parquet to a Volume, COPY INTO" — this PR moves that into the driver so callers don't have to.
How
When `databricks.bulk.volume_path` is set, `executeIngest` switches to:
Geometry handling:
- Walk the schema for `geoarrow.wkb` extension fields. CRS metadata is parsed in two shapes: the simple `{"crs":"EPSG:N"}` string AND DuckDB's `{"crs_type":"projjson","crs":{... "id":{"authority":"EPSG","code":N}}}` PROJJSON form.
- DDL emits `GEOMETRY()` for those columns (bare `GEOMETRY` is rejected on the SQL warehouses I tested).
- The COPY INTO transform projects each through `ST_GEOMFROMWKB(
, <srid_literal>)` so the staged BINARY rebuilds as a typed geometry on insert. The SRID has to be a SQL literal — passing it as a parquet column produces an untyped `GEOMETRY` whose schema can't be merged with the destination column (`DELTA_FAILED_TO_MERGE_FIELDS`).The existing `createTable` / `arrowTypeToDatabricksType` mapping has no geometry case (falls back to BINARY), so the new path uses a sister `createGeoAwareTable` that knows about the detected geo columns. I left the original codepath untouched.
Measurements
End-to-end through DuckDB's `adbc_scanner` calling `adbc_insert(handle, table, (SELECT ...))`. Source = synthetic POINT geometries. Warehouse warmed up before each timed run. Numbers from `tests/databricks-write-bench` and the in-driver harness in duckdb-warehouse-transfer (a CARTO-internal test workspace; happy to inline the harness here if useful).
So the patched driver is ~7,000× faster than the current main at 100 K rows, lands ~81 % of the raw-SQL ceiling, and uses ~30 % of its RAM. RSS is independent of source size because only one Parquet writer + one inflight HTTP body live at a time.
What's not changed
Things I'd like reviewer guidance on
Test plan