You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "XuQianJin-Stars (via GitHub)" <gi...@apache.org> on 2026/02/01 04:14:46 UTC
[PR] [lake/lance] Add Array type support for Lance [fluss]
XuQianJin-Stars opened a new pull request, #2537:
URL: https://github.com/apache/fluss/pull/2537
### Purpose
Linked issue: close #2186
Add Array type support for Lance storage format to enable storing and retrieving tables with array columns.
### Brief change log
**LanceArrowUtils.java:**
- Added ArrayType import and handling in `toArrowType()` method to convert Fluss ArrayType to Arrow List type
- Enhanced `toArrowField()` method to create children fields for ArrayType, properly handling nested array structures by recursively creating element fields
**ArrowDataConverter.java:**
- Added ListVector import for handling Arrow list vectors
- Implemented `copyListVectorData()` method to handle copying of ListVector data between shaded and non-shaded Arrow implementations
- Enhanced `copyVectorData()` method to detect and delegate ListVector copying to the specialized handler
- Added recursive handling for nested array elements in the data conversion process
**LanceTieringTest.java:**
- Added new test case `testTieringWriteTableWithArrayType()` to verify array type support
- Test creates tables with both STRING array and INT array columns
- Test verifies writing and reading array data through Lance storage
- Added helper methods `genArrayTableRecords()` and `verifyArrayTableRecords()` for array data generation and verification
### Tests
- **Unit Test:** `LanceTieringTest#testTieringWriteTableWithArrayType` - Tests writing and reading tables with array columns (STRING array and INT array) through Lance storage
- **Existing Tests:** All existing Lance tiering tests pass, verifying backward compatibility
- **Test Coverage:** Validates array schema conversion, data writing, committing, and reading with proper array element verification
### API and Format
- **No API changes:** This change only adds support for existing ArrayType in Lance storage implementation
- **No format changes:** Uses standard Arrow List format for array storage, which is compatible with Lance
### Documentation
- **No new feature introduction:** This change adds support for an existing Fluss data type (ArrayType) in the Lance storage backend
- **No documentation update needed:** Array type is already documented in Fluss type system; this implementation extends Lance storage to support it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add Array type support for Lance [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #2537:
URL: https://github.com/apache/fluss/pull/2537#discussion_r2753361925
##########
fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringITCase.java:
##########
@@ -144,4 +196,61 @@ private void checkDataInLanceAppendOnlyTable(LanceConfig config, List<InternalRo
assertThat(flussRowIterator.hasNext()).isFalse();
}
}
+
+ private void checkDataInLanceWithAllArrayTypes(
Review Comment:
The data assertion code appears quite verbose and hard to maintain. Could we simplify it by using `readerRoot.contentToTSVString()`—similar to how we assert Flink `Row` results? This would make the test more concise and easier to read.
Additionally, it seems you haven’t merged this test into `testTiering()` yet. Is there a specific reason this can’t be done? If so, could you please clarify?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add Array type support for Lance [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on PR #2537:
URL: https://github.com/apache/fluss/pull/2537#issuecomment-3844827794
@juice928 stop
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add Array type support for Lance [fluss]
Posted by "juice928 (via GitHub)" <gi...@apache.org>.
juice928 commented on PR #2537:
URL: https://github.com/apache/fluss/pull/2537#issuecomment-3842967103
👋 Hi, I'm an automated AI code review bot. I ran some checks on this PR and found **1 point that might be worth attention** (could be false positives, please use your judgment):
1. <strong>Ensure ListVector registers the correct row count</strong>
- [Location](https://github.com/apache/fluss/blob/5f17690fc6aed7cc3db3111d66fd393d54fb1830/fluss-lake/fluss-lake-lance/src/main/java/org/apache/fluss/lake/lance/utils/ArrowDataConverter.java#L140-L144): `fluss-lake/fluss-lake-lance/src/main/java/org/apache/fluss/lake/lance/utils/ArrowDataConverter.java:L140-L144`
- Impact: The destination vector might appear empty to downstream readers because `setLastSet` doesn't automatically update the internal `valueCount`.
- Suggestion: Consider using `setValueCount(valueCount)` instead of `setLastSet` to ensure the vector correctly reflects the amount of data copied.
If you find these suggestions disruptive, you can reply "stop" , and I'll automatically skip this repository in the future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add Array type support for Lance [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong merged PR #2537:
URL: https://github.com/apache/fluss/pull/2537
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add Array type support for Lance [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #2537:
URL: https://github.com/apache/fluss/pull/2537#discussion_r2750734980
##########
fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringTest.java:
##########
@@ -322,4 +325,140 @@ private Schema createTable(LanceConfig config) {
return schema;
}
+
+ @Test
+ void testTieringWriteTableWithArrayType() throws Exception {
Review Comment:
This test looks unusual—it manually instantiates components like `LakeWriter` and `LakeCommitter` using `LanceLakeTieringFactory` to perform tiering. However, this approach is highly error-prone and doesn’t represent the actual end-to-end path that needs to be covered when adding support for a new type like `ARRAY`. Consider this: if we were to add 10 new types, would we really need to write 10 separate, hand-wired integration tests like this? Such tests are not maintainable.
To properly validate `ARRAY` type support, we should perform an end-to-end verification. Specifically, we can simply add a few `ARRAY`-type columns (like `array<string>` and `array<int>`) to the `logTable` created in `org.apache.fluss.lake.lance.tiering.LanceTieringITCase#testTiering`, and then verify that these array fields are correctly written and read in the result assertions.
Additionally, for the Lance format, `ARRAY` types are especially important for **vector embeddings**. Therefore, we should also include a dedicated test case for `ARRAY<FLOAT>` to ensure proper handling of vector data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add Array type support for Lance [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2537:
URL: https://github.com/apache/fluss/pull/2537#discussion_r2750746497
##########
fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringTest.java:
##########
@@ -322,4 +325,140 @@ private Schema createTable(LanceConfig config) {
return schema;
}
+
+ @Test
+ void testTieringWriteTableWithArrayType() throws Exception {
+ int bucketNum = 2;
+ TablePath tablePath = TablePath.of("lance", "arrayTable");
+ Map<String, String> customProperties = new HashMap<>();
+ customProperties.put("lance.batch_size", "256");
+ LanceConfig config =
+ LanceConfig.from(
+ configuration.toMap(),
+ customProperties,
+ tablePath.getDatabaseName(),
+ tablePath.getTableName());
+
+ List<Schema.Column> columns = new ArrayList<>();
+ columns.add(new Schema.Column("id", DataTypes.INT()));
+ columns.add(new Schema.Column("name", DataTypes.STRING()));
+ columns.add(new Schema.Column("tags", DataTypes.ARRAY(DataTypes.STRING())));
+ columns.add(new Schema.Column("scores", DataTypes.ARRAY(DataTypes.INT())));
+ Schema.Builder schemaBuilder = Schema.newBuilder().fromColumns(columns);
+ Schema schema = schemaBuilder.build();
+
+ WriteParams params = LanceConfig.genWriteParamsFromConfig(config);
+ LanceDatasetAdapter.createDataset(
+ config.getDatasetUri(), LanceArrowUtils.toArrowSchema(schema.getRowType()), params);
+
+ TableDescriptor descriptor =
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(bucketNum)
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+ .customProperties(customProperties)
+ .build();
+ TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor, 1L, 1L);
+
+ List<LanceWriteResult> lanceWriteResults = new ArrayList<>();
+ SimpleVersionedSerializer<LanceWriteResult> writeResultSerializer =
+ lanceLakeTieringFactory.getWriteResultSerializer();
+ SimpleVersionedSerializer<LanceCommittable> committableSerializer =
+ lanceLakeTieringFactory.getCommittableSerializer();
+
+ Map<Integer, List<LogRecord>> recordsByBucket = new HashMap<>();
+
+ for (int bucket = 0; bucket < bucketNum; bucket++) {
+ try (LakeWriter<LanceWriteResult> lakeWriter =
+ createLakeWriter(tablePath, bucket, null, tableInfo)) {
+ List<LogRecord> writtenRecords = genArrayTableRecords(bucket, 5);
+ recordsByBucket.put(bucket, writtenRecords);
+ for (LogRecord logRecord : writtenRecords) {
+ lakeWriter.write(logRecord);
+ }
+ LanceWriteResult lanceWriteResult = lakeWriter.complete();
+ byte[] serialized = writeResultSerializer.serialize(lanceWriteResult);
+ lanceWriteResults.add(
+ writeResultSerializer.deserialize(
+ writeResultSerializer.getVersion(), serialized));
+ }
+ }
+
+ try (LakeCommitter<LanceWriteResult, LanceCommittable> lakeCommitter =
+ createLakeCommitter(tablePath, tableInfo)) {
+ LanceCommittable lanceCommittable = lakeCommitter.toCommittable(lanceWriteResults);
+ byte[] serialized = committableSerializer.serialize(lanceCommittable);
+ lanceCommittable =
+ committableSerializer.deserialize(
+ committableSerializer.getVersion(), serialized);
+ Map<String, String> snapshotProperties =
+ Collections.singletonMap(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY, "offsets");
+ long snapshot = lakeCommitter.commit(lanceCommittable, snapshotProperties);
+ assertThat(snapshot).isEqualTo(2);
+ }
+
+ try (Dataset dataset =
+ Dataset.open(
+ new RootAllocator(),
+ config.getDatasetUri(),
+ LanceConfig.genReadOptionFromConfig(config))) {
+ ArrowReader reader = dataset.newScan().scanBatches();
+ VectorSchemaRoot readerRoot = reader.getVectorSchemaRoot();
+
+ for (int bucket = 0; bucket < bucketNum; bucket++) {
+ reader.loadNextBatch();
+ List<LogRecord> expectRecords = recordsByBucket.get(bucket);
+ verifyArrayTableRecords(readerRoot, expectRecords);
+ }
+ assertThat(reader.loadNextBatch()).isFalse();
+ }
+ }
Review Comment:
The test only covers non-null arrays with values. Consider adding test cases for edge cases such as:
1. Null array values (e.g., `genericRow.setField(2, null)`)
2. Empty arrays (e.g., `new GenericArray(new BinaryString[0])`)
3. Nested arrays (e.g., `DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.INT()))`)
These edge cases are important to verify that the Arrow schema conversion and data copying logic handle all scenarios correctly.
##########
fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringTest.java:
##########
@@ -322,4 +325,140 @@ private Schema createTable(LanceConfig config) {
return schema;
}
+
+ @Test
+ void testTieringWriteTableWithArrayType() throws Exception {
+ int bucketNum = 2;
+ TablePath tablePath = TablePath.of("lance", "arrayTable");
+ Map<String, String> customProperties = new HashMap<>();
+ customProperties.put("lance.batch_size", "256");
+ LanceConfig config =
+ LanceConfig.from(
+ configuration.toMap(),
+ customProperties,
+ tablePath.getDatabaseName(),
+ tablePath.getTableName());
+
+ List<Schema.Column> columns = new ArrayList<>();
+ columns.add(new Schema.Column("id", DataTypes.INT()));
+ columns.add(new Schema.Column("name", DataTypes.STRING()));
+ columns.add(new Schema.Column("tags", DataTypes.ARRAY(DataTypes.STRING())));
+ columns.add(new Schema.Column("scores", DataTypes.ARRAY(DataTypes.INT())));
+ Schema.Builder schemaBuilder = Schema.newBuilder().fromColumns(columns);
+ Schema schema = schemaBuilder.build();
+
+ WriteParams params = LanceConfig.genWriteParamsFromConfig(config);
+ LanceDatasetAdapter.createDataset(
+ config.getDatasetUri(), LanceArrowUtils.toArrowSchema(schema.getRowType()), params);
+
+ TableDescriptor descriptor =
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(bucketNum)
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+ .customProperties(customProperties)
+ .build();
+ TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor, 1L, 1L);
+
+ List<LanceWriteResult> lanceWriteResults = new ArrayList<>();
+ SimpleVersionedSerializer<LanceWriteResult> writeResultSerializer =
+ lanceLakeTieringFactory.getWriteResultSerializer();
+ SimpleVersionedSerializer<LanceCommittable> committableSerializer =
+ lanceLakeTieringFactory.getCommittableSerializer();
+
+ Map<Integer, List<LogRecord>> recordsByBucket = new HashMap<>();
+
+ for (int bucket = 0; bucket < bucketNum; bucket++) {
+ try (LakeWriter<LanceWriteResult> lakeWriter =
+ createLakeWriter(tablePath, bucket, null, tableInfo)) {
+ List<LogRecord> writtenRecords = genArrayTableRecords(bucket, 5);
+ recordsByBucket.put(bucket, writtenRecords);
+ for (LogRecord logRecord : writtenRecords) {
+ lakeWriter.write(logRecord);
+ }
+ LanceWriteResult lanceWriteResult = lakeWriter.complete();
+ byte[] serialized = writeResultSerializer.serialize(lanceWriteResult);
+ lanceWriteResults.add(
+ writeResultSerializer.deserialize(
+ writeResultSerializer.getVersion(), serialized));
+ }
+ }
+
+ try (LakeCommitter<LanceWriteResult, LanceCommittable> lakeCommitter =
+ createLakeCommitter(tablePath, tableInfo)) {
+ LanceCommittable lanceCommittable = lakeCommitter.toCommittable(lanceWriteResults);
+ byte[] serialized = committableSerializer.serialize(lanceCommittable);
+ lanceCommittable =
+ committableSerializer.deserialize(
+ committableSerializer.getVersion(), serialized);
+ Map<String, String> snapshotProperties =
+ Collections.singletonMap(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY, "offsets");
+ long snapshot = lakeCommitter.commit(lanceCommittable, snapshotProperties);
+ assertThat(snapshot).isEqualTo(2);
+ }
+
+ try (Dataset dataset =
+ Dataset.open(
+ new RootAllocator(),
+ config.getDatasetUri(),
+ LanceConfig.genReadOptionFromConfig(config))) {
+ ArrowReader reader = dataset.newScan().scanBatches();
+ VectorSchemaRoot readerRoot = reader.getVectorSchemaRoot();
+
+ for (int bucket = 0; bucket < bucketNum; bucket++) {
+ reader.loadNextBatch();
+ List<LogRecord> expectRecords = recordsByBucket.get(bucket);
+ verifyArrayTableRecords(readerRoot, expectRecords);
+ }
+ assertThat(reader.loadNextBatch()).isFalse();
+ }
+ }
+
+ private List<LogRecord> genArrayTableRecords(int bucket, int numRecords) {
+ List<LogRecord> logRecords = new ArrayList<>();
+ for (int i = 0; i < numRecords; i++) {
+ GenericRow genericRow = new GenericRow(4);
+ genericRow.setField(0, i);
+ genericRow.setField(1, BinaryString.fromString("user_" + bucket + "_" + i));
+
+ String[] tagsArray = new String[] {"tag" + i, "tag" + (i + 1), "category" + bucket};
+ BinaryString[] binaryTags = new BinaryString[tagsArray.length];
+ for (int j = 0; j < tagsArray.length; j++) {
+ binaryTags[j] = BinaryString.fromString(tagsArray[j]);
+ }
+ GenericArray tags = new GenericArray(binaryTags);
+ genericRow.setField(2, tags);
+
+ Integer[] scoresArray = new Integer[] {i * 10, i * 20, i * 30};
+ GenericArray scores = new GenericArray(scoresArray);
+ genericRow.setField(3, scores);
+
+ LogRecord logRecord =
+ new GenericRecord(
+ i, System.currentTimeMillis(), ChangeType.APPEND_ONLY, genericRow);
+ logRecords.add(logRecord);
+ }
+ return logRecords;
+ }
+
+ private void verifyArrayTableRecords(VectorSchemaRoot root, List<LogRecord> expectRecords)
+ throws Exception {
+ assertThat(root.getRowCount()).isEqualTo(expectRecords.size());
+ for (int i = 0; i < expectRecords.size(); i++) {
+ LogRecord expectRecord = expectRecords.get(i);
+
+ assertThat((int) (root.getVector(0).getObject(i)))
+ .isEqualTo(expectRecord.getRow().getInt(0));
+ assertThat(((VarCharVector) root.getVector(1)).getObject(i).toString())
+ .isEqualTo(expectRecord.getRow().getString(1).toString());
+
+ ListVector tagsVector = (ListVector) root.getVector(2);
+ Object tagsObject = tagsVector.getObject(i);
+ assertThat(tagsObject).isNotNull();
+
+ ListVector scoresVector = (ListVector) root.getVector(3);
+ Object scoresObject = scoresVector.getObject(i);
+ assertThat(scoresObject).isNotNull();
Review Comment:
The test verification only checks that array objects are not null, but doesn't verify the actual array contents. This means the test would pass even if the arrays are empty or contain incorrect values. The verification should compare the actual array elements against the expected values from the generated records. For example, for the tags array, you should extract the list of strings and compare them with the expected BinaryString values from the expectRecord.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add Array type support for Lance [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #2537:
URL: https://github.com/apache/fluss/pull/2537#discussion_r2751312437
##########
fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringITCase.java:
##########
@@ -94,8 +97,46 @@ void testTiering() throws Exception {
t1.getDatabaseName(),
t1.getTableName());
- // check data in lance
- checkDataInLanceAppendOnlyTable(config, flussRows);
+ // check data in lance, including array columns
+ checkDataInLanceAppendOnlyTableWithArrays(config, flussRows);
+ checkSnapshotPropertyInLance(config, Collections.singletonMap(t1Bucket, 30L));
+
+ jobClient.cancel().get();
+ }
+
+ @Test
+ void testTieringWithFloatArrayForVectorEmbeddings() throws Exception {
Review Comment:
Why not merge this test into `testTiering()` by simply adding multiple array-type columns?
Having a separate test case doesn’t seem to add significant value and unnecessarily increases overall test execution time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add Array type support for Lance [fluss]
Posted by "XuQianJin-Stars (via GitHub)" <gi...@apache.org>.
XuQianJin-Stars commented on PR #2537:
URL: https://github.com/apache/fluss/pull/2537#issuecomment-3833406332
@wuchong @luoyuxia Hi, i already updated the pr. Please help review when you got some time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org