fix: collapse Arrow record batches above configurable cap

src/duckdb/src/table/duckdb-table.ts

@@ -23,6 +23,7 @@ import {
   getApplicationConfig,
   DatabaseAdapter,
   DatabaseConnection,
+  compactArrowTable,
   isArrowTable,
   isArrowVector
 } from '@kepler.gl/utils';
@@ -257,11 +258,13 @@ export class KeplerGlDuckDbTable extends KeplerTable {
 
       restoreGeoarrowMetadata(arrowResult, geoarrowMetadata);
 
+      const compactedResult = compactArrowTable(arrowResult);
+
       fields = useNewFields
-        ? arrowSchemaToFields(arrowResult, tableDuckDBTypes)
-        : data.fields ?? arrowSchemaToFields(arrowResult, tableDuckDBTypes);
-      cols = [...Array(arrowResult.numCols).keys()]
-        .map(i => arrowResult.getChildAt(i))
+        ? arrowSchemaToFields(compactedResult, tableDuckDBTypes)
+        : data.fields ?? arrowSchemaToFields(compactedResult, tableDuckDBTypes);
+      cols = [...Array(compactedResult.numCols).keys()]
+        .map(i => compactedResult.getChildAt(i))
         .filter(col => col) as arrow.Vector[];
     } catch (error) {
       console.error('DuckDB table: createTableAndGetArrow', error);

src/layers/src/geojson-layer/geojson-layer.ts

@@ -617,9 +617,33 @@ export default class GeoJsonLayer extends Layer {
       const geoField = geoFieldAccessor(this.config.columns)(dataContainer);
 
       // update the latest batch/chunk of geoarrow data when loading data incrementally
-      if (geoColumn && geoField && this.dataToFeature.length < dataContainer.numChunks()) {
-        // for incrementally loading data, we only load and render the latest batch; otherwise, we will load and render all batches
-        const isIncrementalLoad = dataContainer.numChunks() - this.dataToFeature.length === 1;
+      const processedRows = this.centroids.length;
+      const numRows = dataContainer.numRows();
+      const numChunks = dataContainer.numChunks();
+      // WKB (and other whole-column collections) keep dataToFeature.length at 1
+      // even when numChunks() is large. Row count is the progress signal then;
+      // comparing length to numChunks would re-parse WKB on every meta update.
+      const processedWholeColumn =
+        this.dataToFeature.length > 0 && processedRows > 0 && processedRows >= numRows;
+      // Progressive loads keep one dataToFeature entry per chunk, then compact
+      // the finished table to one chunk. Row count already matches, so without
+      // this check we would skip the rebuild and keep excess Deck layers.
+      const compactedAfterProgressiveLoad = this.dataToFeature.length > numChunks;
+      const needsUpdate =
+        this.dataToFeature.length === 0 ||
+        (!processedWholeColumn && this.dataToFeature.length < numChunks) ||
+        (processedRows > 0 && processedRows < numRows) ||
+        compactedAfterProgressiveLoad;
+
+      if (geoColumn && geoField && needsUpdate) {
+        // Incremental only when a new chunk appeared. Compacted tables stay at 1
+        // chunk while row count grows, so those updates reprocess the whole table.
+        const isIncrementalLoad =
+          this.dataToFeature.length > 0 &&
+          !processedWholeColumn &&
+          numChunks - this.dataToFeature.length === 1 &&
+          processedRows > 0 &&
+          processedRows < numRows;
         // TODO: add support for COLUMN_MODE_TABLE in getGeojsonLayerMetaFromArrow
         const {dataToFeature, bounds, fixedRadius, featureTypes, centroids} =
           getGeojsonLayerMetaFromArrow({
@@ -628,9 +652,14 @@ export default class GeoJsonLayer extends Layer {
             geoField,
             ...(isIncrementalLoad ? {chunkIndex: this.dataToFeature.length} : null)
           });
-        if (centroids) this.centroids = this.centroids.concat(centroids);
+        if (isIncrementalLoad) {
+          if (centroids) this.centroids = this.centroids.concat(centroids);
+          this.dataToFeature = [...this.dataToFeature, ...dataToFeature];
+        } else {
+          this.centroids = centroids || [];
+          this.dataToFeature = dataToFeature;
+        }
         this.updateMeta({bounds, fixedRadius, featureTypes});
-        this.dataToFeature = [...this.dataToFeature, ...dataToFeature];
       }
     } else if (this.dataToFeature.length === 0 || this.config.columnMode === COLUMN_MODE_TABLE) {
       const getFeature = this.getPositionAccessor(dataContainer);

src/layers/src/heatmap-layer/heatmap-layer.ts

@@ -30,7 +30,7 @@ import {booleanWithin} from '@turf/boolean-within';
 import {point as turfPoint} from '@turf/helpers';
 import {Feature, Polygon} from 'geojson';
 
-import {getGeoArrowPointLayerProps, FindDefaultLayerPropsReturnValue} from '../layer-utils';
+import {getGeoArrowPointLayerProps, FindDefaultLayerPropsReturnValue, getGeoArrowPointCoords} from '../layer-utils';
 import {getFilterDataFunc} from '../aggregation-layer';
 import {
   parseGeoJsonRawFeature,
@@ -88,7 +88,8 @@ export const geoarrowPosAccessor =
   (dc: DataContainerInterface) =>
   (d: {index: number}): number[] => {
     const row = dc.valueAt(d.index, geoarrow.fieldIdx);
-    return [row.get(0), row.get(1)];
+    const coords = getGeoArrowPointCoords(row);
+    return [coords[0], coords[1]];
   };
 
 export const geojsonAccessor =

src/layers/src/layer-utils.ts

@@ -24,14 +24,17 @@ import {convertGeoArrowGeometryToGeoJSON} from '@loaders.gl/gis';
 import {WKBLoader} from '@loaders.gl/wkt';
 import {geojsonToBinary} from '@loaders.gl/gis';
 import {
-  BinaryFeatureCollection,
   Geometry,
   BinaryPointFeature,
   BinaryLineFeature,
   BinaryPolygonFeature
 } from '@loaders.gl/schema';
 
-import {DeckGlGeoTypes, GeojsonDataMaps} from './geojson-layer/geojson-utils';
+import {
+  DeckGlGeoTypes,
+  GeojsonDataMaps,
+  getCentroidFromGeometry
+} from './geojson-layer/geojson-utils';
 
 export type FindDefaultLayerProps = {
   label: string;
@@ -87,7 +90,7 @@ function getBinaryGeometriesFromWKBArrow(
   geoColumn: arrow.Vector,
   options: {chunkIndex?: number; chunkOffset?: number}
 ): GeojsonLayerMetaProps {
-  const dataToFeature: BinaryFeatureCollection[] = [];
+  const centroids: Array<number[] | null> = [];
   const featureTypes: GeojsonLayerMetaProps['featureTypes'] = {
     point: false,
     line: false,
@@ -102,8 +105,8 @@ function getBinaryGeometriesFromWKBArrow(
   let featureIndex = globalFeatureIdOffset;
   let bounds: [number, number, number, number] = [Infinity, Infinity, -Infinity, -Infinity];
 
+  const geojsonFeatures: Feature[] = [];
   chunks.forEach(chunk => {
-    const geojsonFeatures: Feature[] = [];
     for (let i = 0; i < chunk.length; ++i) {
       // ignore features without any geometry
       if (chunk.valueOffsets[i + 1] - chunk.valueOffsets[i] > 0) {
@@ -118,54 +121,57 @@ function getBinaryGeometriesFromWKBArrow(
           properties: {index: featureIndex}
         };
         geojsonFeatures.push(feature);
+        centroids.push(getCentroidFromGeometry(geometry));
 
         const {type} = geometry;
         featureTypes.polygon = type === 'Polygon' || type === 'MultiPolygon';
         featureTypes.point = type === 'Point' || type === 'MultiPoint';
         featureTypes.line = type === 'LineString' || type === 'MultiLineString';
+      } else {
+        centroids.push(null);
       }
 
       featureIndex++;
     }
+  });
 
-    const geojsonToBinaryOptions = {
-      triangulate: true,
-      fixRingWinding: true
-    };
-    const binaryFeatures = geojsonToBinary(geojsonFeatures, geojsonToBinaryOptions);
-
-    // Need to update globalFeatureIds, to take into account previous batches,
-    // as geojsonToBinary doesn't have such option.
-    const featureTypesArr = ['points', 'lines', 'polygons'];
-    featureTypesArr.forEach(prop => {
-      const features = binaryFeatures[prop] as
-        | BinaryPointFeature
-        | BinaryLineFeature
-        | BinaryPolygonFeature;
-      if (features) {
-        bounds = updateBoundsFromGeoArrowSamples(
-          features.positions.value as Float64Array,
-          features.positions.size,
-          bounds
-        );
-
-        const {globalFeatureIds, numericProps} = features;
-        const {index} = numericProps;
-        const len = globalFeatureIds.value.length;
-        for (let i = 0; i < len; ++i) {
-          globalFeatureIds.value[i] = index.value[i];
-        }
+  // One BinaryFeatureCollection for the whole column. A collection per Arrow
+  // chunk becomes one GeoJsonLayer (with fill/stroke/point sublayers) each, and
+  // parquet/DuckDB files easily exceed deck.gl's 255 pickable-layer cap.
+  const geojsonToBinaryOptions = {
+    triangulate: true,
+    fixRingWinding: true
+  };
+  const binaryFeatures = geojsonToBinary(geojsonFeatures, geojsonToBinaryOptions);
+
+  const featureTypesArr = ['points', 'lines', 'polygons'];
+  featureTypesArr.forEach(prop => {
+    const features = binaryFeatures[prop] as
+      | BinaryPointFeature
+      | BinaryLineFeature
+      | BinaryPolygonFeature;
+    if (features) {
+      bounds = updateBoundsFromGeoArrowSamples(
+        features.positions.value as Float64Array,
+        features.positions.size,
+        bounds
+      );
+
+      const {globalFeatureIds, numericProps} = features;
+      const {index} = numericProps;
+      const len = globalFeatureIds.value.length;
+      for (let i = 0; i < len; ++i) {
+        globalFeatureIds.value[i] = index.value[i];
       }
-    });
-
-    dataToFeature.push(binaryFeatures);
+    }
   });
 
   return {
-    dataToFeature: dataToFeature,
+    dataToFeature: [binaryFeatures],
     featureTypes: featureTypes,
     bounds,
-    fixedRadius: false
+    fixedRadius: false,
+    centroids
   };
 }
 
@@ -290,6 +296,38 @@ export function getGeoArrowPointFields(fields: Field[]): Field[] {
   });
 }
 
+/**
+ * Reads lng/lat from a GeoArrow point cell. Arrow may return a FixedSizeList
+ * (with .get), a typed array, a plain [lng, lat] array, or a {x,y} struct.
+ */
+export function getGeoArrowPointCoords(row: any): [number, number, number] {
+  if (row == null) {
+    return [NaN, NaN, 0];
+  }
+  if (typeof row.get === 'function') {
+    const x = row.get(0) ?? row.get('x') ?? row.get('lng') ?? row.get('lon');
+    const y = row.get(1) ?? row.get('y') ?? row.get('lat');
+    if (Number.isFinite(x) && Number.isFinite(y)) {
+      return [Number(x), Number(y), 0];
+    }
+  }
+  if (typeof row.toArray === 'function') {
+    const arr = row.toArray();
+    return [Number(arr[0]), Number(arr[1]), 0];
+  }
+  if (Array.isArray(row) || ArrayBuffer.isView(row)) {
+    return [Number(row[0]), Number(row[1]), 0];
+  }
+  if (typeof row === 'object') {
+    const x = row.x ?? row.lng ?? row.lon;
+    const y = row.y ?? row.lat;
+    if (x !== undefined && y !== undefined) {
+      return [Number(x), Number(y), 0];
+    }
+  }
+  return [NaN, NaN, 0];
+}
+
 /**
  * Builds an arrow vector compatible with ARROW:extension:name geoarrow.point.
  * @param getPosition Position accessor.
@@ -305,7 +343,6 @@ export function createGeoArrowPointVector(
   // in a correct arrow format, as this approach seems too excessive for just a simple interleaved buffer.
 
   const numElements = dataContainer.numRows();
-  const table = dataContainer.getTable();
 
   const numCoords = numElements > 0 ? getPosition({index: 0}).length : 2;
   const precision = 2;
@@ -319,21 +356,16 @@ export function createGeoArrowPointVector(
   const fixedSizeListBuilder = new arrow.FixedSizeListBuilder({type: fixedSizeList});
   fixedSizeListBuilder.addChild(floatBuilder);
 
-  const assembledBatches: arrow.Data[] = [];
+  // One record batch, not one per source batch: a batch per parquet/DuckDB
+  // chunk would create hundreds of pickable ScatterplotLayers and hit deck.gl's
+  // 255-layer picking cap.
   const indexData = {index: 0};
-  for (let batchIndex = 0; batchIndex < table.batches.length; ++batchIndex) {
-    const numRowsInBatch = table.batches[batchIndex].numRows;
-
-    for (let i = 0; i < numRowsInBatch; ++i) {
-      const pos = getPosition(indexData);
-      fixedSizeListBuilder.append(pos);
-
-      ++indexData.index;
-    }
-    assembledBatches.push(fixedSizeListBuilder.flush());
+  for (let i = 0; i < numElements; ++i) {
+    indexData.index = i;
+    fixedSizeListBuilder.append(getPosition(indexData));
   }
 
-  return arrow.makeVector(assembledBatches);
+  return arrow.makeVector(numElements > 0 ? [fixedSizeListBuilder.flush()] : []);
 }
 
 /**

src/layers/src/point-layer/point-layer.ts

@@ -41,7 +41,8 @@ import {
   createGeoArrowPointVector,
   getFilteredIndex,
   getNeighbors,
-  FindDefaultLayerProps
+  FindDefaultLayerProps,
+  getGeoArrowPointCoords
 } from '../layer-utils';
 import {getGeojsonPointDataMaps, GeojsonPointDataMaps} from '../geojson-layer/geojson-utils';
 import {
@@ -131,7 +132,7 @@ export const geoarrowPosAccessor =
   (dataContainer: DataContainerInterface) =>
   (d: {index: number}) => {
     const row = dataContainer.valueAt(d.index, geoarrow.fieldIdx);
-    return [row.get(0), row.get(1), 0];
+    return getGeoArrowPointCoords(row);
   };
 
 export const COLUMN_MODE_POINTS = 'points';

src/processors/src/data-processor.ts

@@ -20,6 +20,7 @@ import {ProcessorResult, Field} from '@kepler.gl/types';
 import {
   arrowDataTypeToAnalyzerDataType,
   arrowDataTypeToFieldType,
+  compactArrowTable,
   hasOwnProperty,
   isPlainObject
 } from '@kepler.gl/utils';
@@ -616,17 +617,34 @@ function castBigIntColumnsToFloat64(arrowTable: arrow.Table): arrow.Table {
   return new arrow.Table(newColumns);
 }
 
+export type ProcessArrowBatchesOptions = {
+  /**
+   * When false, skip compactArrowTable. Progressive Arrow loading uses this so
+   * each incoming batch does not copy the accumulated table. Defaults to true.
+   * The final processFileContent call leaves this unset so the completed table
+   * is compacted once.
+   */
+  compact?: boolean;
+};
+
 /**
  * Parse arrow batches returned from parseInBatches()
  *
- * @param arrowTable the arrow table to parse
+ * @param arrowBatches the arrow record batches to parse
+ * @param options optional processing flags
  * @returns dataset containing `fields` and `rows` or null
  */
-export function processArrowBatches(arrowBatches: arrow.RecordBatch[]): ProcessorResult | null {
+export function processArrowBatches(
+  arrowBatches: arrow.RecordBatch[],
+  options?: ProcessArrowBatchesOptions
+): ProcessorResult | null {
   if (arrowBatches.length === 0) {
     return null;
   }
-  const arrowTable = castBigIntColumnsToFloat64(new arrow.Table(arrowBatches));
+  const arrowTable =
+    options?.compact === false
+      ? castBigIntColumnsToFloat64(new arrow.Table(arrowBatches))
+      : compactArrowTable(castBigIntColumnsToFloat64(new arrow.Table(arrowBatches)));
   const fields = arrowSchemaToFields(arrowTable);
 
   const cols = [...Array(arrowTable.numCols).keys()].map(i => arrowTable.getChildAt(i));

src/processors/src/file-handler.ts

@@ -70,6 +70,11 @@ export type ProcessFileDataContent = {
   sourceUrl?: string;
   /** User-selected or inferred remote file format (csv, geojson, parquet, …). */
   keplerFormat?: string;
+  /**
+   * When true, skip Arrow record-batch compaction. Set on intermediate
+   * progressive Arrow loads so each batch does not copy the growing table.
+   */
+  skipArrowCompact?: boolean;
 };
 
 export {isArrowTable};
@@ -279,7 +284,10 @@ export async function processFileData({
     // eslint-disable-next-line no-useless-catch
     let result;
     try {
-      result = await processor(data);
+      result =
+        format === DATASET_FORMATS.arrow && content.skipArrowCompact
+          ? await processArrowBatches(data as any, {compact: false})
+          : await processor(data);
     } catch (error) {
       throw new Error(`Can not process uploaded file, ${getError(error as Error)}`);
     }