You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "binary-signal (via GitHub)" <gi...@apache.org> on 2025/12/13 12:24:58 UTC
[PR] [lake/paimon] Add ARRAY type support for Fluss-Paimon row conversion [fluss]
binary-signal opened a new pull request, #2166:
URL: https://github.com/apache/fluss/pull/2166
### Purpose
<!-- What is the purpose of the change -->
This PR adds support for ARRAY data types in the Fluss Lake Paimon integration. Previously, attempting to use array types when tiering data to Paimon would throw an `UnsupportedOperationException`. This change enables proper conversion between Fluss and Paimon array types in both directions.
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
- **Added `FlussArrayAsPaimonArray`**: Adapter class that wraps `org.apache.fluss.row.InternalArray` and implements `org.apache.paimon.data.InternalArray`, enabling Fluss arrays to be read as Paimon arrays.
- **Added `PaimonArrayAsFlussArray`**: Adapter class that wraps `org.apache.paimon.data.InternalArray` and implements `org.apache.fluss.row.InternalArray`, enabling Paimon arrays to be read as Fluss arrays.
- **Updated `FlussRowAsPaimonRow.getArray()`**: Now returns a `FlussArrayAsPaimonArray` wrapper instead of throwing `UnsupportedOperationException`.
- **Updated `PaimonRowAsFlussRow.getArray()`**: Now returns a `PaimonArrayAsFlussArray` wrapper instead of throwing `UnsupportedOperationException`.
### Tests
<!-- List UT and IT cases to verify this change -->
**FlussRowAsPaimonRowTest** (new test methods):
- `testArrayTypeWithIntElements()` - Tests array of integers
- `testArrayTypeWithStringElements()` - Tests array of strings
- `testArrayTypeWithNullableElements()` - Tests array with null elements
- `testNullArray()` - Tests null array field
- `testNestedArrayType()` - Tests nested arrays (`ARRAY<ARRAY<INT>>`)
- `testEmptyArray()` - Tests empty array
**PaimonRowAsFlussRowTest** (new test class):
- `testArrayTypeWithIntElements()` - Tests array of integers
- `testArrayTypeWithStringElements()` - Tests array of strings
- `testArrayTypeWithNullableElements()` - Tests array with null elements
- `testNullArray()` - Tests null array field
- `testNestedArrayType()` - Tests nested arrays
- `testEmptyArray()` - Tests empty array
- `testReplaceRow()` - Tests row replacement with arrays
- `testArrayWithLongElements()` - Tests array of longs
- `testArrayWithDoubleElements()` - Tests array of doubles
- `testArrayWithBooleanElements()` - Tests array of booleans
**FlussRecordAsPaimonRowTest** (new test methods):
- `testArrayTypeWithIntElements()` - Tests array with system columns in tiering context
- `testArrayTypeWithStringElements()` - Tests string arrays in tiering context
- `testNestedArrayType()` - Tests nested arrays in tiering context
### API and Format
<!-- Does this change affect API or storage format -->
No. This change is an internal implementation that enables existing array type support to work correctly with Paimon tiering. No public API changes.
### Documentation
<!-- Does this change introduce a new feature -->
No documentation changes required. Array types were already documented as a supported data type; this PR fixes the implementation to actually support them in the Paimon lake integration.
--
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/paimon] Support Array types for tiering paimon [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong merged PR #2166:
URL: https://github.com/apache/fluss/pull/2166
--
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/paimon] Support Array/Nested Array types in tiering service [fluss]
Posted by "binary-signal (via GitHub)" <gi...@apache.org>.
binary-signal commented on code in PR #2166:
URL: https://github.com/apache/fluss/pull/2166#discussion_r2622087438
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/FlussRowAsPaimonRowTest.java:
##########
@@ -140,4 +142,346 @@ void testPrimaryKeyTableRecord() {
assertThat(new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType).getRowKind())
.isEqualTo(RowKind.INSERT);
}
+
+ @Test
+ void testArrayTypeWithIntElements() {
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.IntType(),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.IntType()));
+
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(2);
+ genericRow.setField(0, 42);
+ genericRow.setField(1, new GenericArray(new int[] {1, 2, 3, 4, 5}));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ FlussRowAsPaimonRow flussRowAsPaimonRow =
+ new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType);
+
+ assertThat(flussRowAsPaimonRow.getInt(0)).isEqualTo(42);
+ InternalArray array = flussRowAsPaimonRow.getArray(1);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(5);
+ assertThat(array.getInt(0)).isEqualTo(1);
+ assertThat(array.getInt(1)).isEqualTo(2);
+ assertThat(array.getInt(2)).isEqualTo(3);
+ assertThat(array.getInt(3)).isEqualTo(4);
+ assertThat(array.getInt(4)).isEqualTo(5);
+ }
+
+ @Test
+ void testArrayTypeWithStringElements() {
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.VarCharType(),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.VarCharType()));
+
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(2);
+ genericRow.setField(0, BinaryString.fromString("name"));
+ genericRow.setField(
+ 1,
+ new GenericArray(
+ new Object[] {
+ BinaryString.fromString("a"),
+ BinaryString.fromString("b"),
+ BinaryString.fromString("c")
+ }));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ FlussRowAsPaimonRow flussRowAsPaimonRow =
+ new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType);
+
+ assertThat(flussRowAsPaimonRow.getString(0).toString()).isEqualTo("name");
+ InternalArray array = flussRowAsPaimonRow.getArray(1);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(3);
+ assertThat(array.getString(0).toString()).isEqualTo("a");
+ assertThat(array.getString(1).toString()).isEqualTo("b");
+ assertThat(array.getString(2).toString()).isEqualTo("c");
+ }
+
+ @Test
+ void testArrayTypeWithNullableElements() {
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.IntType().nullable()));
+
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(1);
+ genericRow.setField(0, new GenericArray(new Object[] {1, null, 3}));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ FlussRowAsPaimonRow flussRowAsPaimonRow =
+ new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType);
+
+ InternalArray array = flussRowAsPaimonRow.getArray(0);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(3);
+ assertThat(array.getInt(0)).isEqualTo(1);
+ assertThat(array.isNullAt(1)).isTrue();
+ assertThat(array.getInt(2)).isEqualTo(3);
+ }
+
+ @Test
+ void testNullArray() {
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.ArrayType(new org.apache.paimon.types.IntType())
+ .nullable());
+
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(1);
+ genericRow.setField(0, null);
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ FlussRowAsPaimonRow flussRowAsPaimonRow =
+ new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType);
+
+ assertThat(flussRowAsPaimonRow.isNullAt(0)).isTrue();
+ }
+
+ @Test
+ void testNestedArrayType() {
+ // Test ARRAY<ARRAY<INT>>
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.IntType())));
+
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(1);
+ genericRow.setField(
+ 0,
+ new GenericArray(
+ new Object[] {
+ new GenericArray(new int[] {1, 2}),
+ new GenericArray(new int[] {3, 4, 5})
+ }));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ FlussRowAsPaimonRow flussRowAsPaimonRow =
+ new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType);
+
+ InternalArray outerArray = flussRowAsPaimonRow.getArray(0);
+ assertThat(outerArray).isNotNull();
+ assertThat(outerArray.size()).isEqualTo(2);
+
+ InternalArray innerArray1 = outerArray.getArray(0);
+ assertThat(innerArray1.size()).isEqualTo(2);
+ assertThat(innerArray1.getInt(0)).isEqualTo(1);
+ assertThat(innerArray1.getInt(1)).isEqualTo(2);
+
+ InternalArray innerArray2 = outerArray.getArray(1);
+ assertThat(innerArray2.size()).isEqualTo(3);
+ assertThat(innerArray2.getInt(0)).isEqualTo(3);
+ assertThat(innerArray2.getInt(1)).isEqualTo(4);
+ assertThat(innerArray2.getInt(2)).isEqualTo(5);
+ }
+
+ @Test
+ void testEmptyArray() {
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.IntType()));
+
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(1);
+ genericRow.setField(0, new GenericArray(new int[] {}));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ FlussRowAsPaimonRow flussRowAsPaimonRow =
+ new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType);
+
+ InternalArray array = flussRowAsPaimonRow.getArray(0);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(0);
+ }
+
+ @Test
+ void testArrayWithAllPrimitiveTypes() {
Review Comment:
Yes that make sense I am on 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/paimon] Support Array/Nested Array types in tiering service [fluss]
Posted by "binary-signal (via GitHub)" <gi...@apache.org>.
binary-signal commented on PR #2166:
URL: https://github.com/apache/fluss/pull/2166#issuecomment-3649649708
Adding also 2 PR's related to fluss arrays for visibility
#2165
#2158
--
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/paimon] Support Array/Nested Array types in tiering service [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on PR #2166:
URL: https://github.com/apache/fluss/pull/2166#issuecomment-3678700772
Hi @binary-signal , do you have time to update the pull request? If you're busy, we’d be happy to help with the changes and get it merged.
--
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/paimon] Support Array/Nested Array types in tiering service [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #2166:
URL: https://github.com/apache/fluss/pull/2166#discussion_r2618192592
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/FlussRowAsPaimonRowTest.java:
##########
@@ -140,4 +142,346 @@ void testPrimaryKeyTableRecord() {
assertThat(new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType).getRowKind())
.isEqualTo(RowKind.INSERT);
}
+
+ @Test
+ void testArrayTypeWithIntElements() {
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.IntType(),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.IntType()));
+
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(2);
+ genericRow.setField(0, 42);
+ genericRow.setField(1, new GenericArray(new int[] {1, 2, 3, 4, 5}));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ FlussRowAsPaimonRow flussRowAsPaimonRow =
+ new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType);
+
+ assertThat(flussRowAsPaimonRow.getInt(0)).isEqualTo(42);
+ InternalArray array = flussRowAsPaimonRow.getArray(1);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(5);
+ assertThat(array.getInt(0)).isEqualTo(1);
+ assertThat(array.getInt(1)).isEqualTo(2);
+ assertThat(array.getInt(2)).isEqualTo(3);
+ assertThat(array.getInt(3)).isEqualTo(4);
+ assertThat(array.getInt(4)).isEqualTo(5);
+ }
+
+ @Test
+ void testArrayTypeWithStringElements() {
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.VarCharType(),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.VarCharType()));
+
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(2);
+ genericRow.setField(0, BinaryString.fromString("name"));
+ genericRow.setField(
+ 1,
+ new GenericArray(
+ new Object[] {
+ BinaryString.fromString("a"),
+ BinaryString.fromString("b"),
+ BinaryString.fromString("c")
+ }));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ FlussRowAsPaimonRow flussRowAsPaimonRow =
+ new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType);
+
+ assertThat(flussRowAsPaimonRow.getString(0).toString()).isEqualTo("name");
+ InternalArray array = flussRowAsPaimonRow.getArray(1);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(3);
+ assertThat(array.getString(0).toString()).isEqualTo("a");
+ assertThat(array.getString(1).toString()).isEqualTo("b");
+ assertThat(array.getString(2).toString()).isEqualTo("c");
+ }
+
+ @Test
+ void testArrayTypeWithNullableElements() {
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.IntType().nullable()));
+
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(1);
+ genericRow.setField(0, new GenericArray(new Object[] {1, null, 3}));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ FlussRowAsPaimonRow flussRowAsPaimonRow =
+ new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType);
+
+ InternalArray array = flussRowAsPaimonRow.getArray(0);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(3);
+ assertThat(array.getInt(0)).isEqualTo(1);
+ assertThat(array.isNullAt(1)).isTrue();
+ assertThat(array.getInt(2)).isEqualTo(3);
+ }
+
+ @Test
+ void testNullArray() {
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.ArrayType(new org.apache.paimon.types.IntType())
+ .nullable());
+
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(1);
+ genericRow.setField(0, null);
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ FlussRowAsPaimonRow flussRowAsPaimonRow =
+ new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType);
+
+ assertThat(flussRowAsPaimonRow.isNullAt(0)).isTrue();
+ }
+
+ @Test
+ void testNestedArrayType() {
+ // Test ARRAY<ARRAY<INT>>
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.IntType())));
+
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(1);
+ genericRow.setField(
+ 0,
+ new GenericArray(
+ new Object[] {
+ new GenericArray(new int[] {1, 2}),
+ new GenericArray(new int[] {3, 4, 5})
+ }));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ FlussRowAsPaimonRow flussRowAsPaimonRow =
+ new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType);
+
+ InternalArray outerArray = flussRowAsPaimonRow.getArray(0);
+ assertThat(outerArray).isNotNull();
+ assertThat(outerArray.size()).isEqualTo(2);
+
+ InternalArray innerArray1 = outerArray.getArray(0);
+ assertThat(innerArray1.size()).isEqualTo(2);
+ assertThat(innerArray1.getInt(0)).isEqualTo(1);
+ assertThat(innerArray1.getInt(1)).isEqualTo(2);
+
+ InternalArray innerArray2 = outerArray.getArray(1);
+ assertThat(innerArray2.size()).isEqualTo(3);
+ assertThat(innerArray2.getInt(0)).isEqualTo(3);
+ assertThat(innerArray2.getInt(1)).isEqualTo(4);
+ assertThat(innerArray2.getInt(2)).isEqualTo(5);
+ }
+
+ @Test
+ void testEmptyArray() {
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.IntType()));
+
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(1);
+ genericRow.setField(0, new GenericArray(new int[] {}));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ FlussRowAsPaimonRow flussRowAsPaimonRow =
+ new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType);
+
+ InternalArray array = flussRowAsPaimonRow.getArray(0);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(0);
+ }
+
+ @Test
+ void testArrayWithAllPrimitiveTypes() {
Review Comment:
is it possible to put test for all types in array in this one test method instead having `testArrayWithLongElements`, `testArrayWithDoubleElements`, etc...
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java:
##########
@@ -163,4 +165,384 @@ void testPrimaryKeyTableRecord() {
flussRecordAsPaimonRow.setFlussRecord(logRecord);
assertThat(flussRecordAsPaimonRow.getRowKind()).isEqualTo(RowKind.DELETE);
}
+
+ @Test
+ void testArrayTypeWithIntElements() {
+ int tableBucket = 0;
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.IntType(),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.IntType()),
+ // system columns: __bucket, __offset, __timestamp
+ new org.apache.paimon.types.IntType(),
+ new org.apache.paimon.types.BigIntType(),
+ new org.apache.paimon.types.LocalZonedTimestampType(3));
+
+ FlussRecordAsPaimonRow flussRecordAsPaimonRow =
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ long logOffset = 10;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(2);
+ genericRow.setField(0, 42);
+ genericRow.setField(1, new GenericArray(new int[] {1, 2, 3, 4, 5}));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ flussRecordAsPaimonRow.setFlussRecord(logRecord);
+
+ assertThat(flussRecordAsPaimonRow.getInt(0)).isEqualTo(42);
+ InternalArray array = flussRecordAsPaimonRow.getArray(1);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(5);
+ assertThat(array.getInt(0)).isEqualTo(1);
+ assertThat(array.getInt(1)).isEqualTo(2);
+ assertThat(array.getInt(2)).isEqualTo(3);
+ assertThat(array.getInt(3)).isEqualTo(4);
+ assertThat(array.getInt(4)).isEqualTo(5);
+
+ // Verify system columns are still accessible
+ assertThat(flussRecordAsPaimonRow.getInt(2)).isEqualTo(tableBucket);
+ assertThat(flussRecordAsPaimonRow.getLong(3)).isEqualTo(logOffset);
+ assertThat(flussRecordAsPaimonRow.getLong(4)).isEqualTo(timeStamp);
+ }
+
+ @Test
+ void testArrayTypeWithStringElements() {
+ int tableBucket = 1;
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.VarCharType()),
+ // system columns: __bucket, __offset, __timestamp
+ new org.apache.paimon.types.IntType(),
+ new org.apache.paimon.types.BigIntType(),
+ new org.apache.paimon.types.LocalZonedTimestampType(3));
+
+ FlussRecordAsPaimonRow flussRecordAsPaimonRow =
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ long logOffset = 5;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(1);
+ genericRow.setField(
+ 0,
+ new GenericArray(
+ new Object[] {
+ BinaryString.fromString("hello"), BinaryString.fromString("world")
+ }));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, INSERT, genericRow);
+ flussRecordAsPaimonRow.setFlussRecord(logRecord);
+
+ InternalArray array = flussRecordAsPaimonRow.getArray(0);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(2);
+ assertThat(array.getString(0).toString()).isEqualTo("hello");
+ assertThat(array.getString(1).toString()).isEqualTo("world");
+ }
+
+ @Test
+ void testNestedArrayType() {
+ int tableBucket = 0;
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.IntType())),
+ // system columns: __bucket, __offset, __timestamp
+ new org.apache.paimon.types.IntType(),
+ new org.apache.paimon.types.BigIntType(),
+ new org.apache.paimon.types.LocalZonedTimestampType(3));
+
+ FlussRecordAsPaimonRow flussRecordAsPaimonRow =
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(1);
+ genericRow.setField(
+ 0,
+ new GenericArray(
+ new Object[] {
+ new GenericArray(new int[] {1, 2}),
+ new GenericArray(new int[] {3, 4, 5})
+ }));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ flussRecordAsPaimonRow.setFlussRecord(logRecord);
+
+ InternalArray outerArray = flussRecordAsPaimonRow.getArray(0);
+ assertThat(outerArray).isNotNull();
+ assertThat(outerArray.size()).isEqualTo(2);
+
+ InternalArray innerArray1 = outerArray.getArray(0);
+ assertThat(innerArray1.size()).isEqualTo(2);
+ assertThat(innerArray1.getInt(0)).isEqualTo(1);
+ assertThat(innerArray1.getInt(1)).isEqualTo(2);
+
+ InternalArray innerArray2 = outerArray.getArray(1);
+ assertThat(innerArray2.size()).isEqualTo(3);
+ assertThat(innerArray2.getInt(0)).isEqualTo(3);
+ }
+
+ @Test
+ void testArrayWithAllPrimitiveTypes() {
Review Comment:
dito. Is it possible to test in one single test method instead of introduing many other test method
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/utils/PaimonRowAsFlussRowTest.java:
##########
@@ -0,0 +1,384 @@
+/*
+ * 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.paimon.utils;
+
+import org.apache.fluss.row.InternalArray;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericRow;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link PaimonRowAsFlussRow}. */
+class PaimonRowAsFlussRowTest {
+
+ @Test
+ void testArrayTypeWithIntElements() {
+ // Create a Paimon row with INT and ARRAY<INT> columns
+ // Note: PaimonRowAsFlussRow expects system columns at the end, so we add 3 dummy system
+ // columns
+ GenericRow paimonRow = new GenericRow(4);
+ paimonRow.setField(0, 42);
+ paimonRow.setField(1, new GenericArray(new int[] {1, 2, 3, 4, 5}));
+ // System columns: __bucket, __offset, __timestamp
+ paimonRow.setField(2, 0);
+ paimonRow.setField(3, 0L);
+
+ PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow);
+
+ assertThat(flussRow.getInt(0)).isEqualTo(42);
+ InternalArray array = flussRow.getArray(1);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(5);
+ assertThat(array.getInt(0)).isEqualTo(1);
+ assertThat(array.getInt(1)).isEqualTo(2);
+ assertThat(array.getInt(2)).isEqualTo(3);
+ assertThat(array.getInt(3)).isEqualTo(4);
+ assertThat(array.getInt(4)).isEqualTo(5);
+ }
+
+ @Test
+ void testArrayTypeWithStringElements() {
+ GenericRow paimonRow = new GenericRow(5);
+ paimonRow.setField(0, BinaryString.fromString("name"));
+ paimonRow.setField(
+ 1,
+ new GenericArray(
+ new BinaryString[] {
+ BinaryString.fromString("a"),
+ BinaryString.fromString("b"),
+ BinaryString.fromString("c")
+ }));
+ // System columns: __bucket, __offset, __timestamp
+ paimonRow.setField(2, 0);
+ paimonRow.setField(3, 0L);
+ paimonRow.setField(4, 0L);
+
+ PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow);
+
+ assertThat(flussRow.getString(0).toString()).isEqualTo("name");
+ InternalArray array = flussRow.getArray(1);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(3);
+ assertThat(array.getString(0).toString()).isEqualTo("a");
+ assertThat(array.getString(1).toString()).isEqualTo("b");
+ assertThat(array.getString(2).toString()).isEqualTo("c");
+ }
+
+ @Test
+ void testArrayTypeWithNullableElements() {
+ GenericRow paimonRow = new GenericRow(4);
+ paimonRow.setField(0, new GenericArray(new Object[] {1, null, 3}));
+ // System columns: __bucket, __offset, __timestamp
+ paimonRow.setField(1, 0);
+ paimonRow.setField(2, 0L);
+ paimonRow.setField(3, 0L);
+
+ PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow);
+
+ InternalArray array = flussRow.getArray(0);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(3);
+ assertThat(array.getInt(0)).isEqualTo(1);
+ assertThat(array.isNullAt(1)).isTrue();
+ assertThat(array.getInt(2)).isEqualTo(3);
+ }
+
+ @Test
+ void testNullArray() {
+ GenericRow paimonRow = new GenericRow(4);
+ paimonRow.setField(0, null);
+ // System columns: __bucket, __offset, __timestamp
+ paimonRow.setField(1, 0);
+ paimonRow.setField(2, 0L);
+ paimonRow.setField(3, 0L);
+
+ PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow);
+
+ assertThat(flussRow.isNullAt(0)).isTrue();
+ }
+
+ @Test
+ void testNestedArrayType() {
+ // Test ARRAY<ARRAY<INT>>
+ GenericRow paimonRow = new GenericRow(4);
+ paimonRow.setField(
+ 0,
+ new GenericArray(
+ new Object[] {
+ new GenericArray(new int[] {1, 2}),
+ new GenericArray(new int[] {3, 4, 5})
+ }));
+ // System columns: __bucket, __offset, __timestamp
+ paimonRow.setField(1, 0);
+ paimonRow.setField(2, 0L);
+ paimonRow.setField(3, 0L);
+
+ PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow);
+
+ InternalArray outerArray = flussRow.getArray(0);
+ assertThat(outerArray).isNotNull();
+ assertThat(outerArray.size()).isEqualTo(2);
+
+ InternalArray innerArray1 = outerArray.getArray(0);
+ assertThat(innerArray1.size()).isEqualTo(2);
+ assertThat(innerArray1.getInt(0)).isEqualTo(1);
+ assertThat(innerArray1.getInt(1)).isEqualTo(2);
+
+ InternalArray innerArray2 = outerArray.getArray(1);
+ assertThat(innerArray2.size()).isEqualTo(3);
+ assertThat(innerArray2.getInt(0)).isEqualTo(3);
+ assertThat(innerArray2.getInt(1)).isEqualTo(4);
+ assertThat(innerArray2.getInt(2)).isEqualTo(5);
+ }
+
+ @Test
+ void testEmptyArray() {
+ GenericRow paimonRow = new GenericRow(4);
+ paimonRow.setField(0, new GenericArray(new int[] {}));
+ // System columns: __bucket, __offset, __timestamp
+ paimonRow.setField(1, 0);
+ paimonRow.setField(2, 0L);
+ paimonRow.setField(3, 0L);
+
+ PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow);
+
+ InternalArray array = flussRow.getArray(0);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(0);
+ }
+
+ @Test
+ void testReplaceRow() {
+ GenericRow paimonRow1 = new GenericRow(4);
+ paimonRow1.setField(0, new GenericArray(new int[] {1, 2, 3}));
+ paimonRow1.setField(1, 0);
+ paimonRow1.setField(2, 0L);
+ paimonRow1.setField(3, 0L);
+
+ GenericRow paimonRow2 = new GenericRow(4);
+ paimonRow2.setField(0, new GenericArray(new int[] {4, 5}));
+ paimonRow2.setField(1, 0);
+ paimonRow2.setField(2, 0L);
+ paimonRow2.setField(3, 0L);
+
+ PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow1);
+ assertThat(flussRow.getArray(0).size()).isEqualTo(3);
+
+ flussRow.replaceRow(paimonRow2);
+ assertThat(flussRow.getArray(0).size()).isEqualTo(2);
+ }
+
+ @Test
+ void testArrayWithLongElements() {
+ GenericRow paimonRow = new GenericRow(4);
+ paimonRow.setField(0, new GenericArray(new long[] {100L, 200L, 300L}));
+ paimonRow.setField(1, 0);
+ paimonRow.setField(2, 0L);
+ paimonRow.setField(3, 0L);
+
+ PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow);
+
+ InternalArray array = flussRow.getArray(0);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(3);
+ assertThat(array.getLong(0)).isEqualTo(100L);
+ assertThat(array.getLong(1)).isEqualTo(200L);
+ assertThat(array.getLong(2)).isEqualTo(300L);
+ }
+
+ @Test
+ void testArrayWithDoubleElements() {
+ GenericRow paimonRow = new GenericRow(4);
+ paimonRow.setField(0, new GenericArray(new double[] {1.1, 2.2, 3.3}));
+ paimonRow.setField(1, 0);
+ paimonRow.setField(2, 0L);
+ paimonRow.setField(3, 0L);
+
+ PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow);
+
+ InternalArray array = flussRow.getArray(0);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(3);
+ assertThat(array.getDouble(0)).isEqualTo(1.1);
+ assertThat(array.getDouble(1)).isEqualTo(2.2);
+ assertThat(array.getDouble(2)).isEqualTo(3.3);
+ }
+
+ @Test
+ void testArrayWithBooleanElements() {
+ GenericRow paimonRow = new GenericRow(4);
+ paimonRow.setField(0, new GenericArray(new boolean[] {true, false, true}));
+ paimonRow.setField(1, 0);
+ paimonRow.setField(2, 0L);
+ paimonRow.setField(3, 0L);
+
+ PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow);
+
+ InternalArray array = flussRow.getArray(0);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(3);
+ assertThat(array.getBoolean(0)).isTrue();
+ assertThat(array.getBoolean(1)).isFalse();
+ assertThat(array.getBoolean(2)).isTrue();
+ }
+
+ @Test
+ void testArrayWithAllPrimitiveTypes() {
Review Comment:
dito.
Is it possible to reduce test methods?
--
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/paimon] Support Array/Nested Array types in tiering service [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #2166:
URL: https://github.com/apache/fluss/pull/2166#issuecomment-3690757197
@wuchong Sure. I'll help update the pr
--
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/paimon] Support Array types for tiering paimon [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #2166:
URL: https://github.com/apache/fluss/pull/2166#issuecomment-3691064396
@wuchong 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/paimon] Support Array/Nested Array types in tiering service [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2166:
URL: https://github.com/apache/fluss/pull/2166#discussion_r2646514970
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/utils/PaimonRowAsFlussRowTest.java:
##########
@@ -0,0 +1,198 @@
+/*
+ * 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.paimon.utils;
+
+import org.apache.fluss.row.InternalArray;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link PaimonRowAsFlussRow}. */
+class PaimonRowAsFlussRowTest {
+
+ @Test
+ void testArrayWithAllTypes() {
+ GenericRow paimonRow = new GenericRow(16);
+ // Primitive types
+ paimonRow.setField(0, new GenericArray(new boolean[] {true, false, true}));
+ paimonRow.setField(1, new GenericArray(new byte[] {1, 2, 3}));
+ paimonRow.setField(2, new GenericArray(new short[] {100, 200, 300}));
+ paimonRow.setField(3, new GenericArray(new int[] {1000, 2000, 3000}));
+ paimonRow.setField(4, new GenericArray(new long[] {10000L, 20000L, 30000L}));
+ paimonRow.setField(5, new GenericArray(new float[] {1.1f, 2.2f, 3.3f}));
+ paimonRow.setField(6, new GenericArray(new double[] {1.11, 2.22, 3.33}));
+ // String type
+ paimonRow.setField(
+ 7,
+ new GenericArray(
+ new BinaryString[] {
+ BinaryString.fromString("hello"),
+ BinaryString.fromString("world"),
+ BinaryString.fromString("test")
+ }));
+ // Decimal type
+ paimonRow.setField(
+ 8,
+ new GenericArray(
+ new Object[] {
+ Decimal.fromBigDecimal(new BigDecimal("123.45"), 10, 2),
+ Decimal.fromBigDecimal(new BigDecimal("678.90"), 10, 2),
+ Decimal.fromBigDecimal(new BigDecimal("999.99"), 10, 2)
+ }));
+ // Timestamp type
+ paimonRow.setField(
+ 9,
+ new GenericArray(
+ new Object[] {
+ Timestamp.fromEpochMillis(1698235273182L),
+ Timestamp.fromEpochMillis(1698235274000L),
+ Timestamp.fromEpochMillis(1698235275000L)
+ }));
+ // TimestampLTZ type
+ paimonRow.setField(
+ 10,
+ new GenericArray(
+ new Object[] {
+ Timestamp.fromEpochMillis(1698235273182L),
+ Timestamp.fromEpochMillis(1698235274000L),
+ Timestamp.fromEpochMillis(1698235275000L)
+ }));
+ // Binary type
+ paimonRow.setField(
+ 11,
+ new GenericArray(
+ new Object[] {
+ new byte[] {1, 2, 3},
+ new byte[] {4, 5, 6, 7},
+ new byte[] {8, 9, 10, 11, 12}
+ }));
+ // array<array<int>> type
+ paimonRow.setField(
+ 12,
+ new GenericArray(
+ new Object[] {
+ new GenericArray(new int[] {1, 2}),
+ new GenericArray(new int[] {3, 4, 5})
+ }));
+ // System columns: __bucket, __offset, __timestamp
+ paimonRow.setField(13, 0);
+ paimonRow.setField(14, 0L);
+ paimonRow.setField(15, 0L);
+
+ PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow);
+
+ // Test boolean array
+ InternalArray boolArray = flussRow.getArray(0);
+ assertThat(boolArray.size()).isEqualTo(3);
+ assertThat(boolArray.toBooleanArray()).isEqualTo(new boolean[] {true, false, true});
+
+ // Test byte array
+ InternalArray byteArray = flussRow.getArray(1);
+ assertThat(byteArray.size()).isEqualTo(3);
+ assertThat(byteArray.toByteArray()).isEqualTo(new byte[] {1, 2, 3});
+
+ // Test short array
+ InternalArray shortArray = flussRow.getArray(2);
+ assertThat(shortArray.size()).isEqualTo(3);
+ assertThat(shortArray.toShortArray()).isEqualTo(new short[] {100, 200, 300});
+
+ // Test int array
+ InternalArray intArray = flussRow.getArray(3);
+ assertThat(intArray.size()).isEqualTo(3);
+ assertThat(intArray.toIntArray()).isEqualTo(new int[] {1000, 2000, 3000});
+
+ // Test long array
+ InternalArray longArray = flussRow.getArray(4);
+ assertThat(longArray.size()).isEqualTo(3);
+ assertThat(longArray.toLongArray()).isEqualTo(new long[] {10000L, 20000L, 30000L});
+
+ // Test float array
+ InternalArray floatArray = flussRow.getArray(5);
+ assertThat(floatArray.size()).isEqualTo(3);
+ assertThat(floatArray.toFloatArray()).isEqualTo(new float[] {1.1f, 2.2f, 3.3f});
+
+ // Test double array
+ InternalArray doubleArray = flussRow.getArray(6);
+ assertThat(doubleArray.size()).isEqualTo(3);
+ assertThat(doubleArray.toDoubleArray()).isEqualTo(new double[] {1.11, 2.22, 3.33});
+
+ // Test string array
+ InternalArray stringArray = flussRow.getArray(7);
+ assertThat(stringArray.size()).isEqualTo(3);
+ assertThat(stringArray.getString(0).toString()).isEqualTo("hello");
+ assertThat(stringArray.getString(1).toString()).isEqualTo("world");
+ assertThat(stringArray.getString(2).toString()).isEqualTo("test");
+
+ // Test decimal array
+ InternalArray decimalArray = flussRow.getArray(8);
+ assertThat(decimalArray.size()).isEqualTo(3);
+ assertThat(decimalArray.getDecimal(0, 10, 2).toBigDecimal())
+ .isEqualTo(new BigDecimal("123.45"));
+ assertThat(decimalArray.getDecimal(1, 10, 2).toBigDecimal())
+ .isEqualTo(new BigDecimal("678.90"));
+ assertThat(decimalArray.getDecimal(2, 10, 2).toBigDecimal())
+ .isEqualTo(new BigDecimal("999.99"));
+
+ // Test timestamp array
+ InternalArray timestampArray = flussRow.getArray(9);
+ assertThat(timestampArray.size()).isEqualTo(3);
+ assertThat(timestampArray.getTimestampNtz(0, 3).getMillisecond()).isEqualTo(1698235273182L);
+ assertThat(timestampArray.getTimestampNtz(1, 3).getMillisecond()).isEqualTo(1698235274000L);
+ assertThat(timestampArray.getTimestampNtz(2, 3).getMillisecond()).isEqualTo(1698235275000L);
+
+ // Test timestamp_ltz array
+ InternalArray timestampLtzArray = flussRow.getArray(10);
+ assertThat(timestampLtzArray.size()).isEqualTo(3);
+ assertThat(timestampLtzArray.getTimestampLtz(0, 3).getEpochMillisecond())
+ .isEqualTo(1698235273182L);
+ assertThat(timestampLtzArray.getTimestampLtz(1, 3).getEpochMillisecond())
+ .isEqualTo(1698235274000L);
+ assertThat(timestampLtzArray.getTimestampLtz(2, 3).getEpochMillisecond())
+ .isEqualTo(1698235275000L);
+
+ // Test binary array
+ InternalArray binaryArray = flussRow.getArray(11);
+ assertThat(binaryArray.size()).isEqualTo(3);
+ assertThat(binaryArray.getBinary(0, 3)).isEqualTo(new byte[] {1, 2, 3});
+ assertThat(binaryArray.getBinary(1, 4)).isEqualTo(new byte[] {4, 5, 6, 7});
+ assertThat(binaryArray.getBinary(2, 5)).isEqualTo(new byte[] {8, 9, 10, 11, 12});
+ // Also test getBytes method
+ assertThat(binaryArray.getBytes(0)).isEqualTo(new byte[] {1, 2, 3});
+ assertThat(binaryArray.getBytes(1)).isEqualTo(new byte[] {4, 5, 6, 7});
+ assertThat(binaryArray.getBytes(2)).isEqualTo(new byte[] {8, 9, 10, 11, 12});
+
+ // Also test array<int> array
Review Comment:
The comment says "Also test array<int> array" but should say "Also test array<array<int>>" or "Also test nested array" to accurately describe that it's testing a two-dimensional array type.
```suggestion
// Also test array<array<int>> (nested int array)
```
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/FlussRowAsPaimonRowTest.java:
##########
@@ -140,4 +142,195 @@ void testPrimaryKeyTableRecord() {
assertThat(new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType).getRowKind())
.isEqualTo(RowKind.INSERT);
}
+
+ @Test
+ void testArrayWithAllTypes() {
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.BooleanType()),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.TinyIntType()),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.SmallIntType()),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.IntType()),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.BigIntType()),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.FloatType()),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.DoubleType()),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.VarCharType(true, 30)),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.DecimalType(10, 2)),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.TimestampType(3)),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.LocalZonedTimestampType()),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.VarBinaryType()),
+ // array<array<int>>
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.IntType())));
+
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(13);
+ genericRow.setField(0, new GenericArray(new boolean[] {true, false, true}));
+ genericRow.setField(1, new GenericArray(new byte[] {1, 2, 3}));
+ genericRow.setField(2, new GenericArray(new short[] {100, 200, 300}));
+ genericRow.setField(3, new GenericArray(new Object[] {1000, 2000, 3000}));
+ genericRow.setField(4, new GenericArray(new long[] {10000L, 20000L, 30000L}));
+ genericRow.setField(5, new GenericArray(new float[] {1.1f, 2.2f, 3.3f}));
+ genericRow.setField(6, new GenericArray(new double[] {1.11, 2.22, 3.33}));
+ // String type
+ genericRow.setField(
+ 7,
+ new GenericArray(
+ new BinaryString[] {
+ BinaryString.fromString("hello"),
+ BinaryString.fromString("world"),
+ BinaryString.fromString("test")
+ }));
+
+ // Decimal type
+ genericRow.setField(
+ 8,
+ new GenericArray(
+ new Object[] {
+ Decimal.fromBigDecimal(new BigDecimal("123.45"), 10, 2),
+ Decimal.fromBigDecimal(new BigDecimal("678.90"), 10, 2),
+ Decimal.fromBigDecimal(new BigDecimal("999.99"), 10, 2)
+ }));
+
+ // Timestamp type
+ genericRow.setField(
+ 9,
+ new GenericArray(
+ new Object[] {
+ TimestampNtz.fromMillis(1698235273182L),
+ TimestampNtz.fromMillis(1698235274000L),
+ TimestampNtz.fromMillis(1698235275000L)
+ }));
+
+ // TimestampLTZ type
+ genericRow.setField(
+ 10,
+ new GenericArray(
+ new Object[] {
+ TimestampLtz.fromEpochMillis(1698235273182L),
+ TimestampLtz.fromEpochMillis(1698235274000L),
+ TimestampLtz.fromEpochMillis(1698235275000L)
+ }));
+
+ // Binary type
+ genericRow.setField(
+ 11,
+ new GenericArray(
+ new Object[] {
+ new byte[] {1, 2, 3},
+ new byte[] {4, 5, 6, 7},
+ new byte[] {8, 9, 10, 11, 12}
+ }));
+
+ // array<array<int>> type
+ genericRow.setField(
+ 12,
+ new GenericArray(
+ new Object[] {
+ new GenericArray(new int[] {1, 2}),
+ new GenericArray(new int[] {3, 4, 5})
+ }));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ FlussRowAsPaimonRow flussRowAsPaimonRow =
+ new FlussRowAsPaimonRow(logRecord.getRow(), tableRowType);
+
+ // Test boolean array
+ InternalArray boolArray = flussRowAsPaimonRow.getArray(0);
+ assertThat(boolArray.size()).isEqualTo(3);
+ assertThat(boolArray.toBooleanArray()).isEqualTo(new boolean[] {true, false, true});
+
+ // Test byte array
+ InternalArray byteArray = flussRowAsPaimonRow.getArray(1);
+ assertThat(byteArray.size()).isEqualTo(3);
+ assertThat(byteArray.toByteArray()).isEqualTo(new byte[] {1, 2, 3});
+
+ // Test short array
+ InternalArray shortArray = flussRowAsPaimonRow.getArray(2);
+ assertThat(shortArray.size()).isEqualTo(3);
+ assertThat(shortArray.toShortArray()).isEqualTo(new short[] {100, 200, 300});
+
+ // Test int array
+ InternalArray intArray = flussRowAsPaimonRow.getArray(3);
+ assertThat(intArray.size()).isEqualTo(3);
+ assertThat(intArray.toIntArray()).isEqualTo(new int[] {1000, 2000, 3000});
+
+ // Test long array
+ InternalArray longArray = flussRowAsPaimonRow.getArray(4);
+ assertThat(longArray.size()).isEqualTo(3);
+ assertThat(longArray.toLongArray()).isEqualTo(new long[] {10000L, 20000L, 30000L});
+
+ // Test float array
+ InternalArray floatArray = flussRowAsPaimonRow.getArray(5);
+ assertThat(floatArray.size()).isEqualTo(3);
+ assertThat(floatArray.toFloatArray()).isEqualTo(new float[] {1.1f, 2.2f, 3.3f});
+
+ // Test double array
+ InternalArray doubleArray = flussRowAsPaimonRow.getArray(6);
+ assertThat(doubleArray.size()).isEqualTo(3);
+ assertThat(doubleArray.toDoubleArray()).isEqualTo(new double[] {1.11, 2.22, 3.33});
+
+ // Test string array
+ InternalArray stringArray = flussRowAsPaimonRow.getArray(7);
+ assertThat(stringArray.size()).isEqualTo(3);
+ assertThat(stringArray.getString(0).toString()).isEqualTo("hello");
+ assertThat(stringArray.getString(1).toString()).isEqualTo("world");
+ assertThat(stringArray.getString(2).toString()).isEqualTo("test");
+
+ // Test decimal array
+ InternalArray decimalArray = flussRowAsPaimonRow.getArray(8);
+ assertThat(decimalArray.size()).isEqualTo(3);
+ assertThat(decimalArray.getDecimal(0, 10, 2).toBigDecimal())
+ .isEqualTo(new BigDecimal("123.45"));
+ assertThat(decimalArray.getDecimal(1, 10, 2).toBigDecimal())
+ .isEqualTo(new BigDecimal("678.90"));
+ assertThat(decimalArray.getDecimal(2, 10, 2).toBigDecimal())
+ .isEqualTo(new BigDecimal("999.99"));
+
+ // Test timestamp array
+ InternalArray timestampArray = flussRowAsPaimonRow.getArray(9);
+ assertThat(timestampArray.size()).isEqualTo(3);
+ assertThat(timestampArray.getTimestamp(0, 3).getMillisecond()).isEqualTo(1698235273182L);
+ assertThat(timestampArray.getTimestamp(1, 3).getMillisecond()).isEqualTo(1698235274000L);
+ assertThat(timestampArray.getTimestamp(2, 3).getMillisecond()).isEqualTo(1698235275000L);
+
+ // test timestamp_ltz array
+ timestampArray = flussRowAsPaimonRow.getArray(10);
+ assertThat(timestampArray.size()).isEqualTo(3);
+ assertThat(timestampArray.getTimestamp(0, 3).getMillisecond()).isEqualTo(1698235273182L);
+ assertThat(timestampArray.getTimestamp(1, 3).getMillisecond()).isEqualTo(1698235274000L);
+ assertThat(timestampArray.getTimestamp(2, 3).getMillisecond()).isEqualTo(1698235275000L);
+
+ // Test binary array
+ InternalArray binaryArray = flussRowAsPaimonRow.getArray(11);
+ assertThat(binaryArray.size()).isEqualTo(3);
+ assertThat(binaryArray.getBinary(0)).isEqualTo(new byte[] {1, 2, 3});
+ assertThat(binaryArray.getBinary(1)).isEqualTo(new byte[] {4, 5, 6, 7});
+ assertThat(binaryArray.getBinary(2)).isEqualTo(new byte[] {8, 9, 10, 11, 12});
+
+ // Also test array<int> array
Review Comment:
The comment says "Also test array<int> array" but should say "Also test array<array<int>>" or "Also test nested array" to accurately describe that it's testing a two-dimensional array type.
```suggestion
// Also test array<array<int>> (nested int array)
```
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/FlussArrayAsPaimonArray.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.paimon.source;
+
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.data.InternalMap;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.data.variant.Variant;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataType;
+
+/** Adapter class for converting Fluss InternalArray to Paimon InternalArray. */
+public class FlussArrayAsPaimonArray implements InternalArray {
+
+ private final org.apache.fluss.row.InternalArray flussArray;
+ private final DataType elementType;
+
+ public FlussArrayAsPaimonArray(
+ org.apache.fluss.row.InternalArray flussArray, DataType elementType) {
+ this.flussArray = flussArray;
+ this.elementType = elementType;
+ }
+
+ @Override
+ public int size() {
+ return flussArray.size();
+ }
+
+ @Override
+ public boolean isNullAt(int pos) {
+ return flussArray.isNullAt(pos);
+ }
+
+ @Override
+ public boolean getBoolean(int pos) {
+ return flussArray.getBoolean(pos);
+ }
+
+ @Override
+ public byte getByte(int pos) {
+ return flussArray.getByte(pos);
+ }
+
+ @Override
+ public short getShort(int pos) {
+ return flussArray.getShort(pos);
+ }
+
+ @Override
+ public int getInt(int pos) {
+ return flussArray.getInt(pos);
+ }
+
+ @Override
+ public long getLong(int pos) {
+ return flussArray.getLong(pos);
+ }
+
+ @Override
+ public float getFloat(int pos) {
+ return flussArray.getFloat(pos);
+ }
+
+ @Override
+ public double getDouble(int pos) {
+ return flussArray.getDouble(pos);
+ }
+
+ @Override
+ public BinaryString getString(int pos) {
+ return BinaryString.fromBytes(flussArray.getString(pos).toBytes());
+ }
+
+ @Override
+ public Decimal getDecimal(int pos, int precision, int scale) {
+ org.apache.fluss.row.Decimal flussDecimal = flussArray.getDecimal(pos, precision, scale);
+ if (flussDecimal.isCompact()) {
+ return Decimal.fromUnscaledLong(flussDecimal.toUnscaledLong(), precision, scale);
+ } else {
+ return Decimal.fromBigDecimal(flussDecimal.toBigDecimal(), precision, scale);
+ }
+ }
+
+ @Override
+ public Timestamp getTimestamp(int pos, int precision) {
+ // Default to TIMESTAMP_WITHOUT_TIME_ZONE behavior for arrays
+ switch (elementType.getTypeRoot()) {
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ if (TimestampNtz.isCompact(precision)) {
+ return Timestamp.fromEpochMillis(
+ flussArray.getTimestampNtz(pos, precision).getMillisecond());
+ } else {
+ TimestampNtz timestampNtz = flussArray.getTimestampNtz(pos, precision);
+ return Timestamp.fromEpochMillis(
+ timestampNtz.getMillisecond(), timestampNtz.getNanoOfMillisecond());
+ }
+
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ if (TimestampLtz.isCompact(precision)) {
+ return Timestamp.fromEpochMillis(
+ flussArray.getTimestampLtz(pos, precision).getEpochMillisecond());
+ } else {
+ TimestampLtz timestampLtz = flussArray.getTimestampLtz(pos, precision);
+ return Timestamp.fromEpochMillis(
+ timestampLtz.getEpochMillisecond(),
+ timestampLtz.getNanoOfMillisecond());
+ }
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported data type to get timestamp: " + elementType);
Review Comment:
The getTimestamp method assumes the elementType will always be either TIMESTAMP_WITHOUT_TIME_ZONE or TIMESTAMP_WITH_LOCAL_TIME_ZONE, but doesn't handle other potential timestamp-related types or provide a clear error message for unsupported cases. While the default clause throws an UnsupportedOperationException, the error message could be more specific about which types are supported for arrays.
```suggestion
"Unsupported array element type for getTimestamp. "
+ "Only TIMESTAMP_WITHOUT_TIME_ZONE and "
+ "TIMESTAMP_WITH_LOCAL_TIME_ZONE are supported, but got: "
+ elementType.getTypeRoot()
+ " ("
+ 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/paimon] Add ARRAY type support for Fluss-Paimon row conversion [fluss]
Posted by "binary-signal (via GitHub)" <gi...@apache.org>.
binary-signal commented on PR #2166:
URL: https://github.com/apache/fluss/pull/2166#issuecomment-3649627786
added few mores tests cases hoping coverage will be okay
--
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/paimon] Support Array/Nested Array types in tiering service [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2166:
URL: https://github.com/apache/fluss/pull/2166#discussion_r2618164307
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java:
##########
@@ -163,4 +165,384 @@ void testPrimaryKeyTableRecord() {
flussRecordAsPaimonRow.setFlussRecord(logRecord);
assertThat(flussRecordAsPaimonRow.getRowKind()).isEqualTo(RowKind.DELETE);
}
+
+ @Test
+ void testArrayTypeWithIntElements() {
+ int tableBucket = 0;
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.IntType(),
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.IntType()),
+ // system columns: __bucket, __offset, __timestamp
+ new org.apache.paimon.types.IntType(),
+ new org.apache.paimon.types.BigIntType(),
+ new org.apache.paimon.types.LocalZonedTimestampType(3));
+
+ FlussRecordAsPaimonRow flussRecordAsPaimonRow =
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ long logOffset = 10;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(2);
+ genericRow.setField(0, 42);
+ genericRow.setField(1, new GenericArray(new int[] {1, 2, 3, 4, 5}));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ flussRecordAsPaimonRow.setFlussRecord(logRecord);
+
+ assertThat(flussRecordAsPaimonRow.getInt(0)).isEqualTo(42);
+ InternalArray array = flussRecordAsPaimonRow.getArray(1);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(5);
+ assertThat(array.getInt(0)).isEqualTo(1);
+ assertThat(array.getInt(1)).isEqualTo(2);
+ assertThat(array.getInt(2)).isEqualTo(3);
+ assertThat(array.getInt(3)).isEqualTo(4);
+ assertThat(array.getInt(4)).isEqualTo(5);
+
+ // Verify system columns are still accessible
+ assertThat(flussRecordAsPaimonRow.getInt(2)).isEqualTo(tableBucket);
+ assertThat(flussRecordAsPaimonRow.getLong(3)).isEqualTo(logOffset);
+ assertThat(flussRecordAsPaimonRow.getLong(4)).isEqualTo(timeStamp);
+ }
+
+ @Test
+ void testArrayTypeWithStringElements() {
+ int tableBucket = 1;
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.VarCharType()),
+ // system columns: __bucket, __offset, __timestamp
+ new org.apache.paimon.types.IntType(),
+ new org.apache.paimon.types.BigIntType(),
+ new org.apache.paimon.types.LocalZonedTimestampType(3));
+
+ FlussRecordAsPaimonRow flussRecordAsPaimonRow =
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ long logOffset = 5;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(1);
+ genericRow.setField(
+ 0,
+ new GenericArray(
+ new Object[] {
+ BinaryString.fromString("hello"), BinaryString.fromString("world")
+ }));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, INSERT, genericRow);
+ flussRecordAsPaimonRow.setFlussRecord(logRecord);
+
+ InternalArray array = flussRecordAsPaimonRow.getArray(0);
+ assertThat(array).isNotNull();
+ assertThat(array.size()).isEqualTo(2);
+ assertThat(array.getString(0).toString()).isEqualTo("hello");
+ assertThat(array.getString(1).toString()).isEqualTo("world");
+ }
+
+ @Test
+ void testNestedArrayType() {
+ int tableBucket = 0;
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.ArrayType(
+ new org.apache.paimon.types.IntType())),
+ // system columns: __bucket, __offset, __timestamp
+ new org.apache.paimon.types.IntType(),
+ new org.apache.paimon.types.BigIntType(),
+ new org.apache.paimon.types.LocalZonedTimestampType(3));
+
+ FlussRecordAsPaimonRow flussRecordAsPaimonRow =
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ long logOffset = 0;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(1);
+ genericRow.setField(
+ 0,
+ new GenericArray(
+ new Object[] {
+ new GenericArray(new int[] {1, 2}),
+ new GenericArray(new int[] {3, 4, 5})
+ }));
+
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ flussRecordAsPaimonRow.setFlussRecord(logRecord);
+
+ InternalArray outerArray = flussRecordAsPaimonRow.getArray(0);
+ assertThat(outerArray).isNotNull();
+ assertThat(outerArray.size()).isEqualTo(2);
+
+ InternalArray innerArray1 = outerArray.getArray(0);
+ assertThat(innerArray1.size()).isEqualTo(2);
+ assertThat(innerArray1.getInt(0)).isEqualTo(1);
+ assertThat(innerArray1.getInt(1)).isEqualTo(2);
+
+ InternalArray innerArray2 = outerArray.getArray(1);
+ assertThat(innerArray2.size()).isEqualTo(3);
+ assertThat(innerArray2.getInt(0)).isEqualTo(3);
+ }
Review Comment:
The test assertions for innerArray2 are incomplete. The array contains 3 elements (3, 4, 5) based on the test setup at line 266, but only the first element is verified. The assertions for elements at indices 1 and 2 are missing, making this test less comprehensive than the equivalent tests in PaimonRowAsFlussRowTest and FlussRowAsPaimonRowTest where all elements are verified.
--
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/paimon] Support Array/Nested Array types in tiering service [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on PR #2166:
URL: https://github.com/apache/fluss/pull/2166#issuecomment-3689329464
@luoyuxia could you help to update the PR and I will make a final review.
--
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