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 2025/12/28 01:36:02 UTC
[PR] [lake/iceberg] Support tier array type for iceberg [fluss]
XuQianJin-Stars opened a new pull request, #2266:
URL: https://github.com/apache/fluss/pull/2266
### Purpose
Linked issue: close #2252
This PR adds support for array type conversion between Fluss and Iceberg, enabling tiering of tables with array columns to Iceberg lakehouse storage. The implementation references the Paimon array type support from PR #2166.
### Brief change log
- Updated `FlussDataTypeToIcebergDataType` to convert Fluss `ARRAY` type to Iceberg `LIST` type instead of throwing `UnsupportedOperationException`
- Created `FlussArrayAsIcebergList` adapter class to wrap Fluss `InternalArray` as Java `List` for Iceberg, supporting:
- All primitive types (boolean, byte, short, int, long, float, double)
- Complex types (string, char, decimal, timestamp, timestamp_ltz, date, time, binary, bytes)
- Nested arrays (array of arrays)
- Null handling for both array elements and null arrays
- Updated `FlussRowAsIcebergRecord` to handle array field conversion using the new adapter
- Enhanced `IcebergConversions` to support bidirectional type conversion (Iceberg LIST ↔ Fluss ARRAY)
- Updated documentation to include `ARRAY → LIST` in the Iceberg data type mapping table
### Tests
**Unit Tests** (`FlussRowAsIcebergRecordTest` - 9 test cases):
- `testArrayWithIntElements` - Array of integers
- `testArrayWithStringElements` - Array of strings
- `testNestedArrayType` - Nested arrays (array of arrays)
- `testArrayWithAllPrimitiveTypes` - Arrays of all primitive types
- `testArrayWithDecimalElements` - Array of decimal values
- `testArrayWithTimestampElements` - Arrays of TIMESTAMP and TIMESTAMP_LTZ
- `testArrayWithNullElements` - Arrays with null elements
- `testNullArray` - Null array handling
- `testArrayWithBinaryElements` - Array of binary data
**Integration Tests** (`IcebergTieringTest`):
- `testTieringWriteTableWithArrayType` - Parameterized test (4 cases) covering:
- Primary key tables with array columns (partitioned/non-partitioned)
- Log tables with array columns (partitioned/non-partitioned)
**Test Results**: All 92 tests pass (increased from 88 tests before this PR)
### API and Format
**API Changes**: None. No breaking changes to public APIs.
**Storage Format**: No changes to Fluss storage format. This only affects the conversion layer between Fluss and Iceberg during tiering operations.
**Type Mapping**:
- Fluss `ARRAY<T>` → Iceberg `LIST<T>` (element type `T` is converted recursively)
- Supports nullable and non-nullable arrays
- Maintains element type semantics through recursive conversion
### Documentation
**New Feature**: Yes, this introduces array type support for Iceberg tiering.
**Documentation Changes**:
- Updated `website/docs/streaming-lakehouse/integrate-data-lakes/iceberg.md`
- Added `ARRAY → LIST` mapping to the data type compatibility table
--
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/iceberg] Support tier array type for iceberg [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2266:
URL: https://github.com/apache/fluss/pull/2266#discussion_r2650152599
##########
fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergConversions.java:
##########
@@ -125,6 +125,9 @@ private static DataType convertIcebergTypeToFlussType(Type icebergType) {
} else if (icebergType instanceof Types.DecimalType) {
Types.DecimalType decimalType = (Types.DecimalType) icebergType;
return DataTypes.DECIMAL(decimalType.precision(), decimalType.scale());
+ } else if (icebergType instanceof Types.ListType) {
+ Types.ListType listType = (Types.ListType) icebergType;
+ return DataTypes.ARRAY(convertIcebergTypeToFlussType(listType.elementType()));
Review Comment:
When converting from Iceberg LIST to Fluss ARRAY, the element nullability is not being preserved. The code should check `listType.elementField().isOptional()` and create the element type with the appropriate nullability. If the element field is optional, the element type should be made nullable using `.nullable()`, otherwise it should remain non-nullable.
```suggestion
DataType elementType = convertIcebergTypeToFlussType(listType.elementType());
if (listType.elementField().isOptional()) {
elementType = elementType.nullable();
}
return DataTypes.ARRAY(elementType);
```
##########
fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/FlussDataTypeToIcebergDataType.java:
##########
@@ -129,7 +129,7 @@ public Type visit(LocalZonedTimestampType localZonedTimestampType) {
@Override
public Type visit(ArrayType arrayType) {
- throw new UnsupportedOperationException("Unsupported array type");
+ return Types.ListType.ofOptional(0, arrayType.getElementType().accept(this));
Review Comment:
The code always uses `Types.ListType.ofOptional()`, which creates a list where elements can be null. However, Fluss `ArrayType` can have non-nullable elements. The nullability of the element type should be checked using `arrayType.getElementType().isNullable()` and the appropriate method (`ofRequired` or `ofOptional`) should be used accordingly to preserve the element type's nullability semantics.
```suggestion
Type elementType = arrayType.getElementType().accept(this);
if (arrayType.getElementType().isNullable()) {
return Types.ListType.ofOptional(0, elementType);
} else {
return Types.ListType.ofRequired(0, elementType);
}
```
--
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/iceberg] Support tier array type for iceberg [fluss]
Posted by "XuQianJin-Stars (via GitHub)" <gi...@apache.org>.
XuQianJin-Stars commented on PR #2266:
URL: https://github.com/apache/fluss/pull/2266#issuecomment-3701075204
@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
Re: [PR] [lake/iceberg] Support tier array type for iceberg [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #2266:
URL: https://github.com/apache/fluss/pull/2266
--
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/iceberg] Support tier array type for iceberg [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #2266:
URL: https://github.com/apache/fluss/pull/2266#discussion_r2650241022
##########
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java:
##########
@@ -441,4 +442,256 @@ private void verifyTableRecords(
.isEqualTo(expectRecord.timestamp());
}
}
+
+ @ParameterizedTest
+ @MethodSource("tieringWriteArgs")
+ void testTieringWriteTableWithArrayType(boolean isPrimaryKeyTable, boolean isPartitionedTable)
Review Comment:
move test to `testUnionReadInStreamMode` just like we did for paimon
##########
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/FlussRowAsIcebergRecordTest.java:
##########
@@ -0,0 +1,394 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.iceberg.source;
+
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.Decimal;
+import org.apache.fluss.row.GenericArray;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.types.RowType;
+
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for {@link FlussRowAsIcebergRecord} with array types. */
+class FlussRowAsIcebergRecordTest {
+
+ @Test
+ void testArrayWithIntElements() {
Review Comment:
Combine all test methods to a single test methd to make it trackable?
##########
fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/FlussDataTypeToIcebergDataType.java:
##########
@@ -129,7 +129,7 @@ public Type visit(LocalZonedTimestampType localZonedTimestampType) {
@Override
public Type visit(ArrayType arrayType) {
- throw new UnsupportedOperationException("Unsupported array type");
+ return Types.ListType.ofOptional(0, arrayType.getElementType().accept(this));
Review Comment:
May be we can refer to https://github.com/apache/iceberg/blob/4632f3133efbf6b8232ed9526de7142fd50e1aa6/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java#L47
--
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/iceberg] Support tier array type for iceberg [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #2266:
URL: https://github.com/apache/fluss/pull/2266#discussion_r2659651080
##########
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java:
##########
@@ -441,4 +442,256 @@ private void verifyTableRecords(
.isEqualTo(expectRecord.timestamp());
}
}
+
+ @ParameterizedTest
+ @MethodSource("tieringWriteArgs")
+ void testTieringWriteTableWithArrayType(boolean isPrimaryKeyTable, boolean isPartitionedTable)
Review Comment:
I think we can remove this test to reduce test time cost since we already test the functionality for in `testUnionReadInStreamMode`
--
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