You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "MehulBatra (via GitHub)" <gi...@apache.org> on 2025/07/16 20:55:47 UTC
[PR] [lake/iceberg] Iceberg encoding strategy [fluss]
MehulBatra opened a new pull request, #1350:
URL: https://github.com/apache/fluss/pull/1350
### Purpose
Implement Iceberg encoding strategy for Fluss.
Linked Issue: #1341
### Brief change log
- Add `DataLakeFormat.ICEBERG` enum value
- Implementing `IcebergKeyEncoder` using the binary writer for key encoding
### Tests
None
### API and Format
None
### Documentation
- New feature: Iceberg lake format support for key encoding and bucketing
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on PR #1350:
URL: https://github.com/apache/fluss/pull/1350#issuecomment-3097654330
> > Hi @luoyuxia I tried to take inspiration from https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java for encoding https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Types.java for types
>
> Thanks for the explaination. I think it make senses to be inspired from iceberg's encoding. But keep in mind we use it for different purpose:
>
> * Iceebrg use it to store the binary values of field statistic, right?
> * We use it for distribute records, in Fluss's implementation, it'll first encoding the bucket columns, and then use the encoded value to do bucket partition transfrom which should align with iceberg. But iceberg use literal value to do [bucket partition transfrom](https://github.com/apache/iceberg/blob/6bbbbf6f6f9ba6dc8cb273f8dcd00be1b5dfc399/api/src/main/java/org/apache/iceberg/transforms/Bucket.java#L98). So it means we don't really care about how to encode the bucket columns into binary value. We only need to get the literal value from the binary value(this will be called decoding binary value). That means we can use any encoding way. But It's fine for me to following how iceberg encode the values.
-> Agreed Iceberg does use these to store partition stats and min/max for filtering.
-> The end goal is to encode the values in iceberg compatibal format, can you add more on this?
We only need to get the literal value from the binary value(this will be called decoding binary value). That means we can use any encoding way. But It's fine for me to following how iceberg encode the values.
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2219717551
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
+ * <li>Strings are encoded as UTF-8 bytes without length prefix
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ this.arity = arity;
+ // Conservative initial size to avoid frequent resizing
+ int initialSize = 8 + (arity * 8);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear only the used portion for efficiency
+ if (cursor > 0) {
+ Arrays.fill(buffer, 0, Math.min(cursor, buffer.length), (byte) 0);
+ }
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // For Iceberg key encoding, null values should not occur
+ // This is validated at the encoder level
+ throw new UnsupportedOperationException(
+ "Null values are not supported in Iceberg key encoding");
+ }
+
+ public void writeBoolean(boolean value) {
+ ensureCapacity(1);
+ UnsafeUtils.putBoolean(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeByte(byte value) {
+ ensureCapacity(1);
+ UnsafeUtils.putByte(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(short value) {
+ ensureCapacity(2);
+ UnsafeUtils.putShort(buffer, cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int value) {
+ ensureCapacity(4);
+ UnsafeUtils.putInt(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(long value) {
+ ensureCapacity(8);
+ UnsafeUtils.putLong(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(float value) {
+ ensureCapacity(4);
+ UnsafeUtils.putFloat(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(double value) {
+ ensureCapacity(8);
+ UnsafeUtils.putDouble(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(BinaryString value) {
+ // Based on Iceberg's Conversions.toByteBuffer() for STRING type:
+ // Strings are encoded as UTF-8 bytes without length prefix
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] bytes = BinaryString.encodeUTF8(value.toString());
+ ensureCapacity(bytes.length);
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ void writeBytes(byte[] bytes) {
+ // Based on Iceberg's Conversions.toByteBuffer() for BINARY type:
+ // Binary data is stored directly without length prefix
+ ensureCapacity(bytes.length);
Review Comment:
added comment in the writeDecimal part:
`
public void writeDecimal(Decimal value, int precision) {
byte[] unscaled = value.toUnscaledBytes();
writeBytes(unscaled); // Adds 4-byte length prefix before the actual bytes
}
`
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2216871579
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.record.ChangeType;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/* This file is based on source code of Apache Iceberg Project (https://iceberg.apache.org/), licensed by the Apache
Review Comment:
No, I kept it deliberaty just in case later if i might to copy any iceberg logic, but will remove 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/iceberg] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2218557651
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.record.ChangeType;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/* This file is based on source code of Apache Iceberg Project (https://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding way.
+ *
+ * <p>The logic follows Iceberg's binary encoding format for row data using MemorySegment.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ // Verify MemorySegment uses little-endian for Iceberg compatibility
+ if (!MemorySegment.LITTLE_ENDIAN) {
+ throw new IllegalStateException(
+ "MemorySegment must use little-endian byte order for Iceberg compatibility");
+ }
+
+ this.arity = arity;
+ // Initial buffer size estimation
+ int initialSize = calculateInitialSize(arity);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear buffer
+ Arrays.fill(buffer, 0, Math.min(buffer.length, calculateInitialSize(arity)), (byte) 0);
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // Iceberg handles nulls with a null marker byte
+ ensureCapacity(1);
+ segment.put(cursor, (byte) 0x00); // Null marker
+ cursor += 1;
+ }
+
+ public void writeChangeType(ChangeType kind) {
+ // TODO DISCUSS WITH YUXIA
+ // Iceberg doesn't store ChangeType in binary format like Paimon
+ // This is handled at the metadata/manifest level
+ // We skip this for Iceberg binary encoding
+ }
+
+ public void writeBoolean(int pos, boolean value) {
+ ensureCapacity(1);
+ segment.put(cursor, value ? (byte) 1 : (byte) 0);
+ cursor += 1;
+ }
+
+ public void writeByte(int pos, byte value) {
+ ensureCapacity(1);
+ segment.put(cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(int pos, short value) {
+ ensureCapacity(2);
+ segment.putShort(cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int pos, int value) {
+ ensureCapacity(4);
+ segment.putInt(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(int pos, long value) {
+ ensureCapacity(8);
+ segment.putLong(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(int pos, float value) {
+ ensureCapacity(4);
+ segment.putFloat(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(int pos, double value) {
+ ensureCapacity(8);
+ segment.putDouble(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(int pos, BinaryString input) {
+ if (input == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ byte[] bytes;
+ if (input.getSegments() == null) {
+ bytes = input.toString().getBytes(StandardCharsets.UTF_8);
+ } else {
+ int len = input.getSizeInBytes();
+ bytes = new byte[len];
+ extractFromFlussSegments(input, bytes);
+ }
+ writeBytes(pos, bytes);
+ }
+
+ void writeBytes(int pos, byte[] bytes) {
+ if (bytes == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg variable-length encoding: length prefix + data
+ ensureCapacity(4 + bytes.length);
+ segment.putInt(cursor, bytes.length);
+ cursor += 4;
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ public void writeDecimal(int pos, Decimal value, int precision) {
+ if (value == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg stores decimals as big-endian byte arrays
+ byte[] unscaledBytes = value.toUnscaledBytes();
+
+ // Calculate required bytes based on precision (following Iceberg's rules)
+ int requiredBytes = getIcebergDecimalBytes(precision);
Review Comment:
https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
--
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] Iceberg encoding strategy [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #1350:
URL: https://github.com/apache/fluss/pull/1350#issuecomment-3100291649
> -> The end goal is to encode the values in iceberg compatibal format, can you add more on this?
We don't need to encode the values in iceberg compatibal format. We just need to make sure the bucket partition transform align with iceberg. Iceberg just bucket via literal value, without the encoding process which is different from paimon. Fluss need to encode for we need to align the process that fluss write rows, which will
- step1: first encoding row into bytes
- step2: hash(distribute) the **bytes** into a bucket. So, for iceberg, although we encode it, we still need to decode the bytes into the literal value(a int value, long value, etc), and use the iceberg bucket strategy. See `BucketInteger`, `BucketLong` in iceberg repo
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2221618455
##########
pom.xml:
##########
@@ -98,6 +98,7 @@
<curator.version>5.4.0</curator.version>
<netty.version>4.1.104</netty.version>
<arrow.version>15.0.0</arrow.version>
+ <iceberg.version>1.4.3</iceberg.version>
Review Comment:
```
<!-- paimon bundle, only for test purpose -->
<dependency>
<groupId>org.apache.paimon</groupId>
<artifactId>paimon-bundle</artifactId>
<version>${paimon.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.iceberg</groupId>
<artifactId>iceberg-api</artifactId>
<version>${iceberg.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
```
Paimon and other packages also uses root pom.xml for version, I tried to match and did the same for iceberg-api in fluss-common pom.xml
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2221618455
##########
pom.xml:
##########
@@ -98,6 +98,7 @@
<curator.version>5.4.0</curator.version>
<netty.version>4.1.104</netty.version>
<arrow.version>15.0.0</arrow.version>
+ <iceberg.version>1.4.3</iceberg.version>
Review Comment:
```
<!-- paimon bundle, only for test purpose -->
<dependency>
<groupId>org.apache.paimon</groupId>
<artifactId>paimon-bundle</artifactId>
<version>${paimon.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.iceberg</groupId>
<artifactId>iceberg-api</artifactId>
<version>${iceberg.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
```
Paimon also uses root pom.xml for version, I tried to match and did the same for iceberg-api in fluss-common pom.xml
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2219658617
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
+ * <li>Strings are encoded as UTF-8 bytes without length prefix
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
Review Comment:
Whitelisted, except the UUID as we don't have that datatype in fluss and also iceberg 1.4.3 doesn't support nanoseconds, as it came in table v3, which was released in later version. will add a note for nanoseconds to support post #1195
--
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] Iceberg encoding strategy [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2218690022
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
Review Comment:
We have no UUID, so remove this `UUID types` from the comments.
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
+ * <li>Strings are encoded as UTF-8 bytes without length prefix
Review Comment:
String should encoded with length prefix
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
+ * <li>Strings are encoded as UTF-8 bytes without length prefix
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ this.arity = arity;
+ // Conservative initial size to avoid frequent resizing
+ int initialSize = 8 + (arity * 8);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear only the used portion for efficiency
+ if (cursor > 0) {
+ Arrays.fill(buffer, 0, Math.min(cursor, buffer.length), (byte) 0);
+ }
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // For Iceberg key encoding, null values should not occur
+ // This is validated at the encoder level
+ throw new UnsupportedOperationException(
+ "Null values are not supported in Iceberg key encoding");
+ }
+
+ public void writeBoolean(boolean value) {
+ ensureCapacity(1);
+ UnsafeUtils.putBoolean(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeByte(byte value) {
+ ensureCapacity(1);
+ UnsafeUtils.putByte(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(short value) {
+ ensureCapacity(2);
+ UnsafeUtils.putShort(buffer, cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int value) {
+ ensureCapacity(4);
+ UnsafeUtils.putInt(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(long value) {
+ ensureCapacity(8);
+ UnsafeUtils.putLong(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(float value) {
+ ensureCapacity(4);
+ UnsafeUtils.putFloat(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(double value) {
+ ensureCapacity(8);
+ UnsafeUtils.putDouble(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(BinaryString value) {
+ // Based on Iceberg's Conversions.toByteBuffer() for STRING type:
+ // Strings are encoded as UTF-8 bytes without length prefix
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] bytes = BinaryString.encodeUTF8(value.toString());
+ ensureCapacity(bytes.length);
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ void writeBytes(byte[] bytes) {
+ // Based on Iceberg's Conversions.toByteBuffer() for BINARY type:
+ // Binary data is stored directly without length prefix
+ ensureCapacity(bytes.length);
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ public void writeDecimal(Decimal value, int precision) {
+ // Iceberg stores decimals as unscaled big-endian byte arrays
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] unscaledBytes = value.toUnscaledBytes();
+
+ // For Iceberg, decimals are stored with fixed byte lengths based on precision
+ int requiredBytes = getIcebergDecimalBytes(precision);
+
+ byte[] icebergBytes = new byte[requiredBytes];
+
+ // Convert to big-endian format with proper padding
+ if (unscaledBytes.length <= requiredBytes) {
+ // Pad with sign extension
+ byte paddingByte =
+ (unscaledBytes.length > 0 && (unscaledBytes[0] & 0x80) != 0)
+ ? (byte) 0xFF
+ : (byte) 0x00;
+
+ Arrays.fill(icebergBytes, 0, requiredBytes - unscaledBytes.length, paddingByte);
+ System.arraycopy(
+ unscaledBytes,
+ 0,
+ icebergBytes,
+ requiredBytes - unscaledBytes.length,
+ unscaledBytes.length);
+ } else {
+ // Truncate if too large (should not happen with proper validation)
+ System.arraycopy(unscaledBytes, 0, icebergBytes, 0, requiredBytes);
+ }
+
+ writeBytes(icebergBytes);
+ }
+
+ public void writeTimestampNtz(TimestampNtz value, int precision) {
+ // Iceberg stores timestamps as microseconds since epoch
+ // Reference:
+ // https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ long micros = value.getMillisecond() * 1000L + (value.getNanoOfMillisecond() / 1000L);
+ writeLong(micros);
+ }
+
+ public void writeTimestampLtz(TimestampLtz value, int precision) {
+ // Iceberg stores timestamptz as microseconds since epoch in UTC
+ // Reference:
+ // https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ long epochMillis = value.getEpochMillisecond();
+ int nanoOfMilli = value.getNanoOfMillisecond();
+ long totalMicros = epochMillis * 1000L + (nanoOfMilli / 1000L);
+ writeLong(totalMicros);
+ }
+
+ private void ensureCapacity(int neededSize) {
+ if (buffer.length < cursor + neededSize) {
+ grow(cursor + neededSize);
+ }
+ }
+
+ private void grow(int minCapacity) {
+ int oldCapacity = buffer.length;
+ int newCapacity = oldCapacity + (oldCapacity >> 1); // 1.5x growth
+ if (newCapacity < minCapacity) {
+ newCapacity = minCapacity;
+ }
+ setBuffer(Arrays.copyOf(buffer, newCapacity));
+ }
+
+ private void setBuffer(byte[] buffer) {
+ this.buffer = buffer;
+ this.segment = MemorySegment.wrap(buffer);
+ }
+
+ /**
+ * Returns the number of bytes required to store a decimal with the given precision. Based on
+ * Iceberg's decimal storage specification.
+ *
+ * @param precision the decimal precision
+ * @return number of bytes required
+ */
+ private static int getIcebergDecimalBytes(int precision) {
+ // Reference:
+ // https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ if (precision <= 9) {
+ return 4; // Can fit in 4 bytes
+ } else if (precision <= 18) {
+ return 8; // Can fit in 8 bytes
+ } else {
+ return 16; // Requires 16 bytes
+ }
+ }
+
+ public ByteBuffer toByteBuffer() {
Review Comment:
we don't need this method
##########
pom.xml:
##########
@@ -121,13 +122,13 @@
<spotless.license.header>
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
+ * 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
+ * the License. You may obtain a copy of the License at
Review Comment:
Why change this? Other files are still using two blanks?
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
+ * <li>Strings are encoded as UTF-8 bytes without length prefix
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
Review Comment:
Here is iceberg code for bucket transform, so we only need to support encode the what icberge supports for bucket, for other datatypes, we can just throw unsupport exception to make the code simple
```
switch (type.typeId()) {
case DATE:
case INTEGER:
return (B) new BucketInteger(numBuckets);
case TIME:
case TIMESTAMP:
case LONG:
return (B) new BucketLong(numBuckets);
case DECIMAL:
return (B) new BucketDecimal(numBuckets);
case STRING:
return (B) new BucketString(numBuckets);
case FIXED:
case BINARY:
return (B) new BucketByteBuffer(numBuckets);
case TIMESTAMP_NANO:
return (B) new BucketTimestampNano(numBuckets);
case UUID:
return (B) new BucketUUID(numBuckets);
default:
throw new IllegalArgumentException("Cannot bucket by type: " + type);
}
```
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
+ * <li>Strings are encoded as UTF-8 bytes without length prefix
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ this.arity = arity;
+ // Conservative initial size to avoid frequent resizing
+ int initialSize = 8 + (arity * 8);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear only the used portion for efficiency
+ if (cursor > 0) {
+ Arrays.fill(buffer, 0, Math.min(cursor, buffer.length), (byte) 0);
+ }
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // For Iceberg key encoding, null values should not occur
+ // This is validated at the encoder level
+ throw new UnsupportedOperationException(
+ "Null values are not supported in Iceberg key encoding");
+ }
+
+ public void writeBoolean(boolean value) {
+ ensureCapacity(1);
+ UnsafeUtils.putBoolean(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeByte(byte value) {
+ ensureCapacity(1);
+ UnsafeUtils.putByte(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(short value) {
+ ensureCapacity(2);
+ UnsafeUtils.putShort(buffer, cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int value) {
+ ensureCapacity(4);
+ UnsafeUtils.putInt(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(long value) {
+ ensureCapacity(8);
+ UnsafeUtils.putLong(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(float value) {
+ ensureCapacity(4);
+ UnsafeUtils.putFloat(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(double value) {
+ ensureCapacity(8);
+ UnsafeUtils.putDouble(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(BinaryString value) {
+ // Based on Iceberg's Conversions.toByteBuffer() for STRING type:
+ // Strings are encoded as UTF-8 bytes without length prefix
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] bytes = BinaryString.encodeUTF8(value.toString());
+ ensureCapacity(bytes.length);
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ void writeBytes(byte[] bytes) {
+ // Based on Iceberg's Conversions.toByteBuffer() for BINARY type:
+ // Binary data is stored directly without length prefix
+ ensureCapacity(bytes.length);
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ public void writeDecimal(Decimal value, int precision) {
+ // Iceberg stores decimals as unscaled big-endian byte arrays
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] unscaledBytes = value.toUnscaledBytes();
+
+ // For Iceberg, decimals are stored with fixed byte lengths based on precision
+ int requiredBytes = getIcebergDecimalBytes(precision);
+
+ byte[] icebergBytes = new byte[requiredBytes];
+
+ // Convert to big-endian format with proper padding
+ if (unscaledBytes.length <= requiredBytes) {
+ // Pad with sign extension
+ byte paddingByte =
+ (unscaledBytes.length > 0 && (unscaledBytes[0] & 0x80) != 0)
+ ? (byte) 0xFF
+ : (byte) 0x00;
+
+ Arrays.fill(icebergBytes, 0, requiredBytes - unscaledBytes.length, paddingByte);
+ System.arraycopy(
+ unscaledBytes,
+ 0,
+ icebergBytes,
+ requiredBytes - unscaledBytes.length,
+ unscaledBytes.length);
+ } else {
+ // Truncate if too large (should not happen with proper validation)
+ System.arraycopy(unscaledBytes, 0, icebergBytes, 0, requiredBytes);
+ }
+
+ writeBytes(icebergBytes);
+ }
+
+ public void writeTimestampNtz(TimestampNtz value, int precision) {
+ // Iceberg stores timestamps as microseconds since epoch
+ // Reference:
+ // https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ long micros = value.getMillisecond() * 1000L + (value.getNanoOfMillisecond() / 1000L);
+ writeLong(micros);
+ }
+
+ public void writeTimestampLtz(TimestampLtz value, int precision) {
+ // Iceberg stores timestamptz as microseconds since epoch in UTC
+ // Reference:
Review Comment:
We don't need the Reference comment for every method since we have commented it in header
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
Review Comment:
IIUC, this is not for binary encoding specification for partition We can just remove this line of comment.
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
+ * <li>Strings are encoded as UTF-8 bytes without length prefix
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ this.arity = arity;
+ // Conservative initial size to avoid frequent resizing
+ int initialSize = 8 + (arity * 8);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear only the used portion for efficiency
+ if (cursor > 0) {
+ Arrays.fill(buffer, 0, Math.min(cursor, buffer.length), (byte) 0);
+ }
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // For Iceberg key encoding, null values should not occur
+ // This is validated at the encoder level
+ throw new UnsupportedOperationException(
+ "Null values are not supported in Iceberg key encoding");
+ }
+
+ public void writeBoolean(boolean value) {
+ ensureCapacity(1);
+ UnsafeUtils.putBoolean(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeByte(byte value) {
+ ensureCapacity(1);
+ UnsafeUtils.putByte(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(short value) {
+ ensureCapacity(2);
+ UnsafeUtils.putShort(buffer, cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int value) {
+ ensureCapacity(4);
+ UnsafeUtils.putInt(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(long value) {
+ ensureCapacity(8);
+ UnsafeUtils.putLong(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(float value) {
+ ensureCapacity(4);
+ UnsafeUtils.putFloat(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(double value) {
+ ensureCapacity(8);
+ UnsafeUtils.putDouble(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(BinaryString value) {
+ // Based on Iceberg's Conversions.toByteBuffer() for STRING type:
+ // Strings are encoded as UTF-8 bytes without length prefix
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] bytes = BinaryString.encodeUTF8(value.toString());
+ ensureCapacity(bytes.length);
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ void writeBytes(byte[] bytes) {
+ // Based on Iceberg's Conversions.toByteBuffer() for BINARY type:
+ // Binary data is stored directly without length prefix
+ ensureCapacity(bytes.length);
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ public void writeDecimal(Decimal value, int precision) {
+ // Iceberg stores decimals as unscaled big-endian byte arrays
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] unscaledBytes = value.toUnscaledBytes();
+
+ // For Iceberg, decimals are stored with fixed byte lengths based on precision
+ int requiredBytes = getIcebergDecimalBytes(precision);
+
+ byte[] icebergBytes = new byte[requiredBytes];
+
+ // Convert to big-endian format with proper padding
+ if (unscaledBytes.length <= requiredBytes) {
Review Comment:
Where can i find the complex logic in iceberg. IIUC, iceberg just use `ByteBuffer.wrap(((BigDecimal) value).unscaledValue().toByteArray());`
Can the peice of code simplified?
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
+ * <li>Strings are encoded as UTF-8 bytes without length prefix
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ this.arity = arity;
+ // Conservative initial size to avoid frequent resizing
+ int initialSize = 8 + (arity * 8);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear only the used portion for efficiency
+ if (cursor > 0) {
+ Arrays.fill(buffer, 0, Math.min(cursor, buffer.length), (byte) 0);
+ }
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // For Iceberg key encoding, null values should not occur
+ // This is validated at the encoder level
+ throw new UnsupportedOperationException(
+ "Null values are not supported in Iceberg key encoding");
+ }
+
+ public void writeBoolean(boolean value) {
+ ensureCapacity(1);
+ UnsafeUtils.putBoolean(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeByte(byte value) {
+ ensureCapacity(1);
+ UnsafeUtils.putByte(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(short value) {
+ ensureCapacity(2);
+ UnsafeUtils.putShort(buffer, cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int value) {
+ ensureCapacity(4);
+ UnsafeUtils.putInt(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(long value) {
+ ensureCapacity(8);
+ UnsafeUtils.putLong(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(float value) {
+ ensureCapacity(4);
+ UnsafeUtils.putFloat(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(double value) {
+ ensureCapacity(8);
+ UnsafeUtils.putDouble(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(BinaryString value) {
+ // Based on Iceberg's Conversions.toByteBuffer() for STRING type:
+ // Strings are encoded as UTF-8 bytes without length prefix
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] bytes = BinaryString.encodeUTF8(value.toString());
Review Comment:
We need to put the length of the bytes.
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
+ * <li>Strings are encoded as UTF-8 bytes without length prefix
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ this.arity = arity;
+ // Conservative initial size to avoid frequent resizing
+ int initialSize = 8 + (arity * 8);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear only the used portion for efficiency
+ if (cursor > 0) {
+ Arrays.fill(buffer, 0, Math.min(cursor, buffer.length), (byte) 0);
+ }
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // For Iceberg key encoding, null values should not occur
+ // This is validated at the encoder level
+ throw new UnsupportedOperationException(
+ "Null values are not supported in Iceberg key encoding");
+ }
+
+ public void writeBoolean(boolean value) {
+ ensureCapacity(1);
+ UnsafeUtils.putBoolean(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeByte(byte value) {
+ ensureCapacity(1);
+ UnsafeUtils.putByte(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(short value) {
+ ensureCapacity(2);
+ UnsafeUtils.putShort(buffer, cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int value) {
+ ensureCapacity(4);
+ UnsafeUtils.putInt(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(long value) {
+ ensureCapacity(8);
+ UnsafeUtils.putLong(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(float value) {
+ ensureCapacity(4);
+ UnsafeUtils.putFloat(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(double value) {
+ ensureCapacity(8);
+ UnsafeUtils.putDouble(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(BinaryString value) {
+ // Based on Iceberg's Conversions.toByteBuffer() for STRING type:
+ // Strings are encoded as UTF-8 bytes without length prefix
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] bytes = BinaryString.encodeUTF8(value.toString());
+ ensureCapacity(bytes.length);
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ void writeBytes(byte[] bytes) {
+ // Based on Iceberg's Conversions.toByteBuffer() for BINARY type:
+ // Binary data is stored directly without length prefix
+ ensureCapacity(bytes.length);
Review Comment:
We also need to put the length of the bytes.
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergKeyEncoder.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.encode.KeyEncoder;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.types.RowType;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import static com.alibaba.fluss.utils.Preconditions.checkArgument;
+
+/** An implementation of {@link KeyEncoder} to follow Iceberg's encoding strategy. */
+public class IcebergKeyEncoder implements KeyEncoder {
+
+ private final InternalRow.FieldGetter[] fieldGetters;
+
+ private final IcebergBinaryRowWriter.FieldWriter[] fieldEncoders;
+
+ private final IcebergBinaryRowWriter icebergBinaryRowWriter;
+
+ public IcebergKeyEncoder(RowType rowType, List<String> keys) {
+
+ // Validate single key field requirement as per FIP
+ checkArgument(
+ keys.size() == 1,
+ "Key fields must have exactly one field for iceberg format, but got: %s",
+ keys.size());
+
+ // for get fields from fluss internal row
+ fieldGetters = new InternalRow.FieldGetter[keys.size()];
+ // for encode fields into iceberg
+ fieldEncoders = new IcebergBinaryRowWriter.FieldWriter[keys.size()];
+ for (int i = 0; i < keys.size(); i++) {
+ int keyIndex = rowType.getFieldIndex(keys.get(i));
+ DataType keyDataType = rowType.getTypeAt(keyIndex);
+ fieldGetters[i] = InternalRow.createFieldGetter(keyDataType, keyIndex);
+ fieldEncoders[i] = IcebergBinaryRowWriter.createFieldWriter(keyDataType);
+ }
+
+ icebergBinaryRowWriter = new IcebergBinaryRowWriter(keys.size());
+ }
+
+ @Override
+ public byte[] encodeKey(InternalRow row) {
+ icebergBinaryRowWriter.reset();
+ // iterate all the fields of the row, and encode each field
+ for (int i = 0; i < fieldGetters.length; i++) {
+ fieldEncoders[i].writeField(
+ icebergBinaryRowWriter, fieldGetters[i].getFieldOrNull(row));
+ }
+ return icebergBinaryRowWriter.toBytes();
+ }
+
+ /**
+ * Returns the encoded key as a ByteBuffer for Iceberg-compatible operations. This is useful
+ * when interfacing with Iceberg's hash functions which expect ByteBuffer.
+ */
+ public ByteBuffer encodeKeyAsBuffer(InternalRow row) {
Review Comment:
We don't need this method.
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
Review Comment:
I think we will still need length prefix for varlength type, such as string. That's different from iceberg. If no length prefix, we can't figure out the bound of string values if we store two or more string values.
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergKeyEncoder.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.encode.KeyEncoder;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.types.RowType;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import static com.alibaba.fluss.utils.Preconditions.checkArgument;
+
+/** An implementation of {@link KeyEncoder} to follow Iceberg's encoding strategy. */
+public class IcebergKeyEncoder implements KeyEncoder {
+
+ private final InternalRow.FieldGetter[] fieldGetters;
+
+ private final IcebergBinaryRowWriter.FieldWriter[] fieldEncoders;
+
+ private final IcebergBinaryRowWriter icebergBinaryRowWriter;
+
+ public IcebergKeyEncoder(RowType rowType, List<String> keys) {
+
+ // Validate single key field requirement as per FIP
+ checkArgument(
+ keys.size() == 1,
+ "Key fields must have exactly one field for iceberg format, but got: %s",
+ keys.size());
Review Comment:
```suggestion
keys);
```
it will be better to print the keys
##########
pom.xml:
##########
@@ -550,7 +551,7 @@
under the License.
-->
<license implementation="org.apache.rat.analysis.license.SimplePatternBasedLicense">
- <licenseFamilyCategory>AL2 </licenseFamilyCategory>
+ <licenseFamilyCategory>AL2</licenseFamilyCategory>
Review Comment:
dito
##########
pom.xml:
##########
@@ -121,13 +122,13 @@
<spotless.license.header>
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
+ * 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
+ * the License. You may obtain a copy of the License at
*
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
Review Comment:
dito
##########
pom.xml:
##########
@@ -832,10 +833,10 @@
<transformers combine.children="append">
<!-- The service transformer is needed to merge META-INF/services files -->
<transformer
- implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
Review Comment:
dito
##########
fluss-common/src/test/java/com/alibaba/fluss/row/encode/iceberg/IcebergKeyEncoderTest.java:
##########
@@ -0,0 +1,431 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.row.indexed.IndexedRow;
+import com.alibaba.fluss.row.indexed.IndexedRowWriter;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.types.DataTypes;
+import com.alibaba.fluss.types.RowType;
+
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Unit tests for {@link IcebergKeyEncoder} to verify the encoding matches Iceberg's format.
+ *
+ * <p>This test uses Iceberg's actual Conversions class to ensure our encoding is byte-for-byte
+ * compatible with Iceberg's implementation.
+ */
+class IcebergKeyEncoderTest {
+
+ @Test
+ void testSingleKeyFieldRequirement() {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.STRING()},
+ new String[] {"id", "name"});
+
+ // Should succeed with single key
+ IcebergKeyEncoder encoder = new IcebergKeyEncoder(rowType, Collections.singletonList("id"));
+ assertThat(encoder).isNotNull();
+
+ // Should fail with multiple keys
+ assertThatThrownBy(() -> new IcebergKeyEncoder(rowType, Arrays.asList("id", "name")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Key fields must have exactly one field");
+ }
+
+ @Test
+ void testIntegerEncoding() {
+ RowType rowType = RowType.of(new DataType[] {DataTypes.INT()}, new String[] {"id"});
+
+ int testValue = 42;
+ IndexedRow row = createRowWithInt(testValue);
+ IcebergKeyEncoder encoder = new IcebergKeyEncoder(rowType, Collections.singletonList("id"));
+
+ // Encode with our implementation
+ byte[] ourEncoded = encoder.encodeKey(row);
+
+ // Encode with Iceberg's implementation
+ ByteBuffer icebergBuffer = Conversions.toByteBuffer(Types.IntegerType.get(), testValue);
+ byte[] icebergEncoded = toByteArray(icebergBuffer);
+
+ assertThat(ourEncoded).isEqualTo(icebergEncoded);
+ }
+
+ @Test
+ void testLongEncoding() {
+ RowType rowType = RowType.of(new DataType[] {DataTypes.BIGINT()}, new String[] {"id"});
+
+ long testValue = 1234567890123456789L;
+ IndexedRow row = createRowWithLong(testValue);
+ IcebergKeyEncoder encoder = new IcebergKeyEncoder(rowType, Collections.singletonList("id"));
+
+ // Encode with our implementation
+ byte[] ourEncoded = encoder.encodeKey(row);
+
+ // Encode with Iceberg's implementation
+ ByteBuffer icebergBuffer = Conversions.toByteBuffer(Types.LongType.get(), testValue);
+ byte[] icebergEncoded = toByteArray(icebergBuffer);
+
+ assertThat(ourEncoded).isEqualTo(icebergEncoded);
+ }
+
+ @Test
+ void testStringEncoding() {
+ RowType rowType = RowType.of(new DataType[] {DataTypes.STRING()}, new String[] {"name"});
+
+ String testValue = "Hello Iceberg, Fluss this side!";
+ IndexedRow row = createRowWithString(testValue);
+ IcebergKeyEncoder encoder =
+ new IcebergKeyEncoder(rowType, Collections.singletonList("name"));
+
+ // Encode with our implementation
+ byte[] ourEncoded = encoder.encodeKey(row);
+
+ // Encode with Iceberg's implementation
+ ByteBuffer icebergBuffer = Conversions.toByteBuffer(Types.StringType.get(), testValue);
+ byte[] icebergEncoded = toByteArray(icebergBuffer);
+
+ assertThat(ourEncoded).isEqualTo(icebergEncoded);
+ }
+
+ @Test
+ void testBooleanEncoding() {
+ RowType rowType = RowType.of(new DataType[] {DataTypes.BOOLEAN()}, new String[] {"flag"});
+
+ // Test true
+ IndexedRow rowTrue = createRowWithBoolean(true);
+ IcebergKeyEncoder encoder =
+ new IcebergKeyEncoder(rowType, Collections.singletonList("flag"));
+
+ byte[] ourEncodedTrue = encoder.encodeKey(rowTrue);
+ ByteBuffer icebergBufferTrue = Conversions.toByteBuffer(Types.BooleanType.get(), true);
+ byte[] icebergEncodedTrue = toByteArray(icebergBufferTrue);
+
+ assertThat(ourEncodedTrue).isEqualTo(icebergEncodedTrue);
+
+ // Test false
+ IndexedRow rowFalse = createRowWithBoolean(false);
+ byte[] ourEncodedFalse = encoder.encodeKey(rowFalse);
+ ByteBuffer icebergBufferFalse = Conversions.toByteBuffer(Types.BooleanType.get(), false);
+ byte[] icebergEncodedFalse = toByteArray(icebergBufferFalse);
+
+ assertThat(ourEncodedFalse).isEqualTo(icebergEncodedFalse);
+ }
+
+ @Test
+ void testFloatDoubleEncoding() {
+ // Test float
+ RowType floatRowType =
+ RowType.of(new DataType[] {DataTypes.FLOAT()}, new String[] {"value"});
+
+ float floatVal = 3.14f;
+ IndexedRow floatRow = createRowWithFloat(floatVal);
+ IcebergKeyEncoder floatEncoder =
+ new IcebergKeyEncoder(floatRowType, Collections.singletonList("value"));
+
+ byte[] ourEncodedFloat = floatEncoder.encodeKey(floatRow);
+ ByteBuffer icebergBufferFloat = Conversions.toByteBuffer(Types.FloatType.get(), floatVal);
+ byte[] icebergEncodedFloat = toByteArray(icebergBufferFloat);
+
+ assertThat(ourEncodedFloat).isEqualTo(icebergEncodedFloat);
+
+ // Test double
+ RowType doubleRowType =
+ RowType.of(new DataType[] {DataTypes.DOUBLE()}, new String[] {"value"});
+
+ double doubleVal = 3.14159265359;
+ IndexedRow doubleRow = createRowWithDouble(doubleVal);
+ IcebergKeyEncoder doubleEncoder =
+ new IcebergKeyEncoder(doubleRowType, Collections.singletonList("value"));
+
+ byte[] ourEncodedDouble = doubleEncoder.encodeKey(doubleRow);
+ ByteBuffer icebergBufferDouble =
+ Conversions.toByteBuffer(Types.DoubleType.get(), doubleVal);
+ byte[] icebergEncodedDouble = toByteArray(icebergBufferDouble);
+
+ assertThat(ourEncodedDouble).isEqualTo(icebergEncodedDouble);
+ }
+
+ @Test
+ void testDecimalEncoding() {
+ RowType rowType =
+ RowType.of(new DataType[] {DataTypes.DECIMAL(10, 2)}, new String[] {"amount"});
+
+ BigDecimal testValue = new BigDecimal("123.45");
+ IndexedRow row = createRowWithDecimal(testValue, 10, 2);
+ IcebergKeyEncoder encoder =
+ new IcebergKeyEncoder(rowType, Collections.singletonList("amount"));
+
+ // Encode with our implementation
+ byte[] ourEncoded = encoder.encodeKey(row);
+
+ // Encode with Iceberg's implementation
+ Type.PrimitiveType decimalType = Types.DecimalType.of(10, 2);
+ ByteBuffer icebergBuffer = Conversions.toByteBuffer(decimalType, testValue);
+ byte[] icebergEncoded = toByteArray(icebergBuffer);
+
+ assertThat(ourEncoded).isEqualTo(icebergEncoded);
+ }
+
+ @Test
+ void testTimestampEncoding() {
+ RowType rowType =
+ RowType.of(new DataType[] {DataTypes.TIMESTAMP(6)}, new String[] {"event_time"});
+
+ // Iceberg expects microseconds for TIMESTAMP type
+ long millis = 1698235273182L;
+ int nanos = 123456;
+ long micros = millis * 1000 + (nanos / 1000);
+
+ IndexedRow row = createRowWithTimestampNtz(millis, nanos);
+ IcebergKeyEncoder encoder =
+ new IcebergKeyEncoder(rowType, Collections.singletonList("event_time"));
+
+ // Encode with our implementation
+ byte[] ourEncoded = encoder.encodeKey(row);
+
+ // Encode with Iceberg's implementation
+ ByteBuffer icebergBuffer =
+ Conversions.toByteBuffer(Types.TimestampType.withoutZone(), micros);
+ byte[] icebergEncoded = toByteArray(icebergBuffer);
+
+ assertThat(ourEncoded).isEqualTo(icebergEncoded);
+ }
+
+ @Test
+ void testTimestampWithTimezoneEncoding() {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.TIMESTAMP_LTZ(6)}, new String[] {"event_time"});
+
+ // Iceberg expects microseconds for TIMESTAMP type
+ long millis = 1698235273182L;
+ int nanos = 45678;
+ long micros = millis * 1000 + (nanos / 1000);
+
+ IndexedRow row = createRowWithTimestampLtz(millis, nanos);
+ IcebergKeyEncoder encoder =
+ new IcebergKeyEncoder(rowType, Collections.singletonList("event_time"));
+
+ // Encode with our implementation
+ byte[] ourEncoded = encoder.encodeKey(row);
+
+ // Encode with Iceberg's implementation
+ ByteBuffer icebergBuffer = Conversions.toByteBuffer(Types.TimestampType.withZone(), micros);
+ byte[] icebergEncoded = toByteArray(icebergBuffer);
+
+ assertThat(ourEncoded).isEqualTo(icebergEncoded);
+ }
+
+ @Test
+ void testDateEncoding() {
+ RowType rowType = RowType.of(new DataType[] {DataTypes.DATE()}, new String[] {"date"});
+
+ // Date value as days since epoch
+ int dateValue = 19655; // 2023-10-25
+ IndexedRow row = createRowWithDate(dateValue);
+ IcebergKeyEncoder encoder =
+ new IcebergKeyEncoder(rowType, Collections.singletonList("date"));
+
+ // Encode with our implementation
+ byte[] ourEncoded = encoder.encodeKey(row);
+
+ // Encode with Iceberg's implementation
+ ByteBuffer icebergBuffer = Conversions.toByteBuffer(Types.DateType.get(), dateValue);
+ byte[] icebergEncoded = toByteArray(icebergBuffer);
+
+ assertThat(ourEncoded).isEqualTo(icebergEncoded);
+ }
+
+ @Test
+ void testTimeEncoding() {
+ RowType rowType = RowType.of(new DataType[] {DataTypes.TIME()}, new String[] {"time"});
+
+ // Fluss stores time as int (milliseconds since midnight)
+ int timeMillis = 34200000;
+ long timeMicros = timeMillis * 1000L; // Convert to microseconds for Iceberg
+
+ IndexedRow row = createRowWithTime(timeMillis);
+ IcebergKeyEncoder encoder =
+ new IcebergKeyEncoder(rowType, Collections.singletonList("time"));
+
+ // Encode with our implementation
+ byte[] ourEncoded = encoder.encodeKey(row);
+
+ // Encode with Iceberg's implementation (expects microseconds as long)
+ ByteBuffer icebergBuffer = Conversions.toByteBuffer(Types.TimeType.get(), timeMicros);
+ byte[] icebergEncoded = toByteArray(icebergBuffer);
+
+ assertThat(ourEncoded).isEqualTo(icebergEncoded);
+ }
+
+ @Test
+ void testBinaryEncoding() {
+ RowType rowType = RowType.of(new DataType[] {DataTypes.BYTES()}, new String[] {"data"});
+
+ byte[] testValue = "Hello i only understand binary data".getBytes();
+ IndexedRow row = createRowWithBytes(testValue);
+ IcebergKeyEncoder encoder =
+ new IcebergKeyEncoder(rowType, Collections.singletonList("data"));
+
+ // Encode with our implementation
+ byte[] ourEncoded = encoder.encodeKey(row);
+
+ // Encode with Iceberg's implementation
+ // Iceberg expects ByteBuffer for BINARY type
+ ByteBuffer icebergBuffer =
+ Conversions.toByteBuffer(Types.BinaryType.get(), ByteBuffer.wrap(testValue));
+ byte[] icebergEncoded = toByteArray(icebergBuffer);
+
+ assertThat(ourEncoded).isEqualTo(icebergEncoded);
+ }
+
+ // Helper method to convert ByteBuffer to byte array
+ private byte[] toByteArray(ByteBuffer buffer) {
+ byte[] array = new byte[buffer.remaining()];
+ buffer.get(array);
+ return array;
+ }
+
+ // ---- Helper methods to create IndexedRow instances ----
+
+ private IndexedRow createRowWithInt(int value) {
+ DataType[] dataTypes = {DataTypes.INT()};
+ IndexedRowWriter writer = new IndexedRowWriter(dataTypes);
+ writer.writeInt(value);
+ IndexedRow row = new IndexedRow(dataTypes);
+ row.pointTo(writer.segment(), 0, writer.position());
+ return row;
+ }
+
+ private IndexedRow createRowWithLong(long value) {
+ DataType[] dataTypes = {DataTypes.BIGINT()};
+ IndexedRowWriter writer = new IndexedRowWriter(dataTypes);
+ writer.writeLong(value);
+ IndexedRow row = new IndexedRow(dataTypes);
+ row.pointTo(writer.segment(), 0, writer.position());
+ return row;
+ }
+
+ private IndexedRow createRowWithString(String value) {
+ DataType[] dataTypes = {DataTypes.STRING()};
+ IndexedRowWriter writer = new IndexedRowWriter(dataTypes);
+ writer.writeString(BinaryString.fromString(value));
+ IndexedRow row = new IndexedRow(dataTypes);
+ row.pointTo(writer.segment(), 0, writer.position());
+ return row;
+ }
+
+ private IndexedRow createRowWithBoolean(boolean value) {
+ DataType[] dataTypes = {DataTypes.BOOLEAN()};
+ IndexedRowWriter writer = new IndexedRowWriter(dataTypes);
+ writer.writeBoolean(value);
+ IndexedRow row = new IndexedRow(dataTypes);
+ row.pointTo(writer.segment(), 0, writer.position());
+ return row;
+ }
+
+ private IndexedRow createRowWithFloat(float value) {
+ DataType[] dataTypes = {DataTypes.FLOAT()};
+ IndexedRowWriter writer = new IndexedRowWriter(dataTypes);
+ writer.writeFloat(value);
+ IndexedRow row = new IndexedRow(dataTypes);
+ row.pointTo(writer.segment(), 0, writer.position());
+ return row;
+ }
+
+ private IndexedRow createRowWithDouble(double value) {
+ DataType[] dataTypes = {DataTypes.DOUBLE()};
+ IndexedRowWriter writer = new IndexedRowWriter(dataTypes);
+ writer.writeDouble(value);
+ IndexedRow row = new IndexedRow(dataTypes);
+ row.pointTo(writer.segment(), 0, writer.position());
+ return row;
+ }
+
+ private IndexedRow createRowWithDecimal(BigDecimal value, int precision, int scale) {
+ DataType[] dataTypes = {DataTypes.DECIMAL(precision, scale)};
+ IndexedRowWriter writer = new IndexedRowWriter(dataTypes);
+ writer.writeDecimal(Decimal.fromBigDecimal(value, precision, scale), precision);
+ IndexedRow row = new IndexedRow(dataTypes);
+ row.pointTo(writer.segment(), 0, writer.position());
+ return row;
+ }
+
+ private IndexedRow createRowWithTimestampNtz(long millis, int nanos) {
+ DataType[] dataTypes = {DataTypes.TIMESTAMP(6)};
+ IndexedRowWriter writer = new IndexedRowWriter(dataTypes);
+ writer.writeTimestampNtz(TimestampNtz.fromMillis(millis, nanos), 6);
+ IndexedRow row = new IndexedRow(dataTypes);
+ row.pointTo(writer.segment(), 0, writer.position());
+ return row;
+ }
+
+ private IndexedRow createRowWithTimestampLtz(long millis, int nanos) {
+ DataType[] dataTypes = {DataTypes.TIMESTAMP_LTZ(6)};
+ IndexedRowWriter writer = new IndexedRowWriter(dataTypes);
+ writer.writeTimestampLtz(TimestampLtz.fromEpochMillis(millis, nanos), 6);
+ IndexedRow row = new IndexedRow(dataTypes);
+ row.pointTo(writer.segment(), 0, writer.position());
+ return row;
+ }
+
+ private IndexedRow createRowWithDate(int days) {
+ DataType[] dataTypes = {DataTypes.DATE()};
+ IndexedRowWriter writer = new IndexedRowWriter(dataTypes);
+ writer.writeInt(days);
+ IndexedRow row = new IndexedRow(dataTypes);
+ row.pointTo(writer.segment(), 0, writer.position());
+ return row;
+ }
+
+ private IndexedRow createRowWithTime(int millis) {
+ DataType[] dataTypes = {DataTypes.TIME()};
+ IndexedRowWriter writer = new IndexedRowWriter(dataTypes);
+ writer.writeInt(millis); // Fluss stores TIME as int (milliseconds)
+ IndexedRow row = new IndexedRow(dataTypes);
+ row.pointTo(writer.segment(), 0, writer.position());
+ return row;
+ }
+
+ private IndexedRow createRowWithBytes(byte[] value) {
Review Comment:
nit:
```
private IndexedRow createRowWithBytes(byte[] value) throws Exception {
DataType[] dataTypes = {DataTypes.BYTES()};
try (IndexedRowWriter writer = new IndexedRowWriter(dataTypes)) {
writer.writeBytes(value);
IndexedRow row = new IndexedRow(dataTypes);
row.pointTo(writer.segment(), 0, writer.position());
return row;
}
}
```
Close `IndexedRowWriter` to resolve the warning of IDE
##########
pom.xml:
##########
@@ -832,10 +833,10 @@
<transformers combine.children="append">
<!-- The service transformer is needed to merge META-INF/services files -->
<transformer
- implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
<!-- The ApacheNoticeResourceTransformer collects and aggregates NOTICE files -->
<transformer
- implementation="org.apache.maven.plugins.shade.resource.ApacheNoticeResourceTransformer">
+ implementation="org.apache.maven.plugins.shade.resource.ApacheNoticeResourceTransformer">
Review Comment:
dito
--
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] Iceberg encoding strategy [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #1350:
URL: https://github.com/apache/fluss/pull/1350#issuecomment-3095965537
> Hi @luoyuxia I tried to take inspiration from https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java for encoding https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Types.java for types
Thanks for the explaination. I think it make senses to be inspired from iceberg's encoding. But keep in mind we use it for different purpose:
- Iceebrg use it to store the binary values of field statistic, right?
- We use it for distribute records, in Fluss's implementation, it'll first encoding the bucket columns, and then use the encoded value to do bucket partition transfrom which should align with iceberg. But iceberg use literal value to do [bucket partition transfrom](https://github.com/apache/iceberg/blob/6bbbbf6f6f9ba6dc8cb273f8dcd00be1b5dfc399/api/src/main/java/org/apache/iceberg/transforms/Bucket.java#L98). So it means we don't really care about how to encode the bucket columns into binary value. We only need to get the literal value from the binary value(this will be called decoding binary value). That means we can use any encoding way. But It's fine for me to following how iceberg encode the values.
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2219601983
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
Review Comment:
Bit confusing in support types comment for iceberg buckettransform you mentioned to provide support for UUID, here it's said remove it, so to be on the safer side i am removing UUID as we don't have support for that datatype in fluss
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2217290321
##########
fluss-common/src/test/java/com/alibaba/fluss/row/encode/iceberg/IcebergKeyEncoderTest.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.row.indexed.IndexedRow;
+import com.alibaba.fluss.row.indexed.IndexedRowWriter;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.types.DataTypes;
+import com.alibaba.fluss.types.RowType;
+import com.alibaba.fluss.utils.TypeUtils;
+
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.List;
+
+import static com.alibaba.fluss.row.TestInternalRowGenerator.createAllRowType;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** UT for {@link IcebergKeyEncoder} to verify the encoding result matches Iceberg's format. */
+class IcebergKeyEncoderTest {
+
+ @Test
+ void testEncodeKey() {
+ // create a row with all types
+ RowType allRowType = createAllRowType();
+ DataType[] allDataTypes = allRowType.getChildren().toArray(new DataType[0]);
+
+ IndexedRow indexedRow = genFlussRowForAllTypes(allDataTypes);
+ List<String> encodedKeys = allRowType.getFieldNames();
+ IcebergKeyEncoder icebergKeyEncoder = new IcebergKeyEncoder(allRowType, encodedKeys);
+
+ // encode with Fluss own implementation for Iceberg
+ byte[] encodedKey = icebergKeyEncoder.encodeKey(indexedRow);
+
+ // encode with Iceberg implementation using native Iceberg classes
Review Comment:
Sure I will see how can I encode the rows using iceberg library itself.
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2219710567
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
+ * <li>Strings are encoded as UTF-8 bytes without length prefix
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ this.arity = arity;
+ // Conservative initial size to avoid frequent resizing
+ int initialSize = 8 + (arity * 8);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear only the used portion for efficiency
+ if (cursor > 0) {
+ Arrays.fill(buffer, 0, Math.min(cursor, buffer.length), (byte) 0);
+ }
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // For Iceberg key encoding, null values should not occur
+ // This is validated at the encoder level
+ throw new UnsupportedOperationException(
+ "Null values are not supported in Iceberg key encoding");
+ }
+
+ public void writeBoolean(boolean value) {
+ ensureCapacity(1);
+ UnsafeUtils.putBoolean(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeByte(byte value) {
+ ensureCapacity(1);
+ UnsafeUtils.putByte(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(short value) {
+ ensureCapacity(2);
+ UnsafeUtils.putShort(buffer, cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int value) {
+ ensureCapacity(4);
+ UnsafeUtils.putInt(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(long value) {
+ ensureCapacity(8);
+ UnsafeUtils.putLong(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(float value) {
+ ensureCapacity(4);
+ UnsafeUtils.putFloat(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(double value) {
+ ensureCapacity(8);
+ UnsafeUtils.putDouble(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(BinaryString value) {
+ // Based on Iceberg's Conversions.toByteBuffer() for STRING type:
+ // Strings are encoded as UTF-8 bytes without length prefix
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] bytes = BinaryString.encodeUTF8(value.toString());
+ ensureCapacity(bytes.length);
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ void writeBytes(byte[] bytes) {
+ // Based on Iceberg's Conversions.toByteBuffer() for BINARY type:
+ // Binary data is stored directly without length prefix
+ ensureCapacity(bytes.length);
Review Comment:
added length, in this also, but this is adding lenght prefix in decimal also.
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
+ * <li>Strings are encoded as UTF-8 bytes without length prefix
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ this.arity = arity;
+ // Conservative initial size to avoid frequent resizing
+ int initialSize = 8 + (arity * 8);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear only the used portion for efficiency
+ if (cursor > 0) {
+ Arrays.fill(buffer, 0, Math.min(cursor, buffer.length), (byte) 0);
+ }
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // For Iceberg key encoding, null values should not occur
+ // This is validated at the encoder level
+ throw new UnsupportedOperationException(
+ "Null values are not supported in Iceberg key encoding");
+ }
+
+ public void writeBoolean(boolean value) {
+ ensureCapacity(1);
+ UnsafeUtils.putBoolean(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeByte(byte value) {
+ ensureCapacity(1);
+ UnsafeUtils.putByte(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(short value) {
+ ensureCapacity(2);
+ UnsafeUtils.putShort(buffer, cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int value) {
+ ensureCapacity(4);
+ UnsafeUtils.putInt(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(long value) {
+ ensureCapacity(8);
+ UnsafeUtils.putLong(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(float value) {
+ ensureCapacity(4);
+ UnsafeUtils.putFloat(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(double value) {
+ ensureCapacity(8);
+ UnsafeUtils.putDouble(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(BinaryString value) {
+ // Based on Iceberg's Conversions.toByteBuffer() for STRING type:
+ // Strings are encoded as UTF-8 bytes without length prefix
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] bytes = BinaryString.encodeUTF8(value.toString());
+ ensureCapacity(bytes.length);
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ void writeBytes(byte[] bytes) {
+ // Based on Iceberg's Conversions.toByteBuffer() for BINARY type:
+ // Binary data is stored directly without length prefix
+ ensureCapacity(bytes.length);
Review Comment:
added length, in this also, but this is adding lenght prefix in writedecimal also.
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2219493213
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
+ * <li>Strings are encoded as UTF-8 bytes without length prefix
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ this.arity = arity;
+ // Conservative initial size to avoid frequent resizing
+ int initialSize = 8 + (arity * 8);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear only the used portion for efficiency
+ if (cursor > 0) {
+ Arrays.fill(buffer, 0, Math.min(cursor, buffer.length), (byte) 0);
+ }
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // For Iceberg key encoding, null values should not occur
+ // This is validated at the encoder level
+ throw new UnsupportedOperationException(
+ "Null values are not supported in Iceberg key encoding");
+ }
+
+ public void writeBoolean(boolean value) {
+ ensureCapacity(1);
+ UnsafeUtils.putBoolean(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeByte(byte value) {
+ ensureCapacity(1);
+ UnsafeUtils.putByte(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(short value) {
+ ensureCapacity(2);
+ UnsafeUtils.putShort(buffer, cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int value) {
+ ensureCapacity(4);
+ UnsafeUtils.putInt(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(long value) {
+ ensureCapacity(8);
+ UnsafeUtils.putLong(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(float value) {
+ ensureCapacity(4);
+ UnsafeUtils.putFloat(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(double value) {
+ ensureCapacity(8);
+ UnsafeUtils.putDouble(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(BinaryString value) {
+ // Based on Iceberg's Conversions.toByteBuffer() for STRING type:
+ // Strings are encoded as UTF-8 bytes without length prefix
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] bytes = BinaryString.encodeUTF8(value.toString());
+ ensureCapacity(bytes.length);
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ void writeBytes(byte[] bytes) {
+ // Based on Iceberg's Conversions.toByteBuffer() for BINARY type:
+ // Binary data is stored directly without length prefix
+ ensureCapacity(bytes.length);
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ public void writeDecimal(Decimal value, int precision) {
+ // Iceberg stores decimals as unscaled big-endian byte arrays
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] unscaledBytes = value.toUnscaledBytes();
+
+ // For Iceberg, decimals are stored with fixed byte lengths based on precision
+ int requiredBytes = getIcebergDecimalBytes(precision);
+
+ byte[] icebergBytes = new byte[requiredBytes];
+
+ // Convert to big-endian format with proper padding
+ if (unscaledBytes.length <= requiredBytes) {
Review Comment:
While reffering to doc, it was little complicated, but where as in the conversion code it's easy, I have translated it to a simpler way now.
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2219668920
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
Review Comment:
make sense
--
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] Iceberg encoding strategy [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2221895963
##########
pom.xml:
##########
@@ -98,6 +98,7 @@
<curator.version>5.4.0</curator.version>
<netty.version>4.1.104</netty.version>
<arrow.version>15.0.0</arrow.version>
+ <iceberg.version>1.4.3</iceberg.version>
Review Comment:
Sorry for missing that test need it. I'd like to retract this comment
--
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] Iceberg encoding strategy [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #1350:
URL: https://github.com/apache/fluss/pull/1350
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2219670306
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
Review Comment:
ok, i will try to encode the lenght alongside
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2219781155
##########
pom.xml:
##########
@@ -832,10 +833,10 @@
<transformers combine.children="append">
<!-- The service transformer is needed to merge META-INF/services files -->
<transformer
- implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
Review Comment:
Somehow the formatter is not letting me change it back, it is getting autofixed
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on PR #1350:
URL: https://github.com/apache/fluss/pull/1350#issuecomment-3095766909
Hi @luoyuxia I tried to take inspiration from
https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
for encoding
https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Types.java
for types
--
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] Iceberg encoding strategy [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2220751937
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
Review Comment:
If we have two string columns, if not encode length of each string column, how can we know the bound of the bytes of the string column?
--
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] Iceberg encoding strategy [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2220762719
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link com.alibaba.fluss.row.InternalRow} using Iceberg's binary
+ * encoding format.
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal types use BIG-ENDIAN byte order
+ * <li>Strings are encoded as UTF-8 bytes
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ this.arity = arity;
+ // Conservative initial size to avoid frequent resizing
+ int initialSize = 8 + (arity * 8);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear only the used portion for efficiency
+ if (cursor > 0) {
+ Arrays.fill(buffer, 0, Math.min(cursor, buffer.length), (byte) 0);
+ }
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // For Iceberg key encoding, null values should not occur
+ // This is validated at the encoder level
+ throw new UnsupportedOperationException(
+ "Null values are not supported in Iceberg key encoding");
+ }
+
+ // todo: Revisit to add support as per Iceberg 1.9.1 post #1195 merge for Java 11 support
Review Comment:
Why put the todo in here? I suggest to add the todo in https://github.com/apache/fluss/pull/1350/files#diff-6051cc48685fc3e562c38ddbede5594cadce48e06db5b359c57e9c37aefb6ca6L36
##########
pom.xml:
##########
@@ -98,6 +98,7 @@
<curator.version>5.4.0</curator.version>
<netty.version>4.1.104</netty.version>
<arrow.version>15.0.0</arrow.version>
+ <iceberg.version>1.4.3</iceberg.version>
Review Comment:
remove this, we dont need it in root `pom.xml`
##########
fluss-common/src/test/java/com/alibaba/fluss/row/encode/iceberg/IcebergKeyEncoderTest.java:
##########
@@ -0,0 +1,395 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.row.indexed.IndexedRow;
+import com.alibaba.fluss.row.indexed.IndexedRowWriter;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.types.DataTypes;
+import com.alibaba.fluss.types.RowType;
+
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.Collections;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Unit tests for {@link IcebergKeyEncoder} to verify the encoding matches Iceberg's format.
+ *
+ * <p>This test uses Iceberg's actual Conversions class to ensure our encoding is byte-for-byte
+ * compatible with Iceberg's implementation.
+ */
+class IcebergKeyEncoderTest {
+
+ // todo: Revisit to add support as per Iceberg 1.9.1 post #1195 merge for Java 11 support
Review Comment:
dito: remove the todo in here.
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on PR #1350:
URL: https://github.com/apache/fluss/pull/1350#issuecomment-3095945968
Currently our CI runs for java8, when I am using iceberg 1.9.1 in my code as it only supports java 11 , it is failing the CI check, trying to downgrade the version to 1.4.3 to support java 8, my assumption it would pass post that.
Stacktrace:
`Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.8.0:testCompile (default-testCompile) on project fluss-common: Compilation failure
Error: /home/fluss/actions-runner/_work/fluss/fluss/fluss-common/src/test/java/com/alibaba/fluss/row/encode/iceberg/IcebergKeyEncoderTest.java:[30,32] cannot access org.apache.iceberg.types.Conversions
Error: bad class file: /home/fluss/.m2/repository/org/apache/iceberg/iceberg-api/1.9.1/iceberg-api-1.9.1.jar(org/apache/iceberg/types/Conversions.class)
Error: class file has wrong version 55.0, should be 52.0
Error: Please remove or make sure it appears in the correct subdirectory of the classpath.`
--
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] Iceberg encoding strategy [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #1350:
URL: https://github.com/apache/fluss/pull/1350#issuecomment-3096023161
> Currently our CI runs for java8, when I am using iceberg 1.9.1 in my code as it only supports java 11 , it is failing the CI check, trying to downgrade the version to 1.4.3 to support java 8, my assumption it would pass post that. Stacktrace: `Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.8.0:testCompile (default-testCompile) on project fluss-common: Compilation failure Error: /home/fluss/actions-runner/_work/fluss/fluss/fluss-common/src/test/java/com/alibaba/fluss/row/encode/iceberg/IcebergKeyEncoderTest.java:[30,32] cannot access org.apache.iceberg.types.Conversions Error: bad class file: /home/fluss/.m2/repository/org/apache/iceberg/iceberg-api/1.9.1/iceberg-api-1.9.1.jar(org/apache/iceberg/types/Conversions.class) Error: class file has wrong version 55.0, should be 52.0 Error: Please remove or make sure it appears in the correct subdirectory of the classpath.`
If that's the case, we can downgrade the version to 1.4.3 firstly, and left todo when #1195 is resolved
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2219219181
##########
pom.xml:
##########
@@ -121,13 +122,13 @@
<spotless.license.header>
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
+ * 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
+ * the License. You may obtain a copy of the License at
Review Comment:
my ide formatter doing something unsual will fix 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/iceberg] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2219496958
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
+ * <li>Strings are encoded as UTF-8 bytes without length prefix
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ this.arity = arity;
+ // Conservative initial size to avoid frequent resizing
+ int initialSize = 8 + (arity * 8);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear only the used portion for efficiency
+ if (cursor > 0) {
+ Arrays.fill(buffer, 0, Math.min(cursor, buffer.length), (byte) 0);
+ }
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // For Iceberg key encoding, null values should not occur
+ // This is validated at the encoder level
+ throw new UnsupportedOperationException(
+ "Null values are not supported in Iceberg key encoding");
+ }
+
+ public void writeBoolean(boolean value) {
+ ensureCapacity(1);
+ UnsafeUtils.putBoolean(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeByte(byte value) {
+ ensureCapacity(1);
+ UnsafeUtils.putByte(buffer, cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(short value) {
+ ensureCapacity(2);
+ UnsafeUtils.putShort(buffer, cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int value) {
+ ensureCapacity(4);
+ UnsafeUtils.putInt(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(long value) {
+ ensureCapacity(8);
+ UnsafeUtils.putLong(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(float value) {
+ ensureCapacity(4);
+ UnsafeUtils.putFloat(buffer, cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(double value) {
+ ensureCapacity(8);
+ UnsafeUtils.putDouble(buffer, cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(BinaryString value) {
+ // Based on Iceberg's Conversions.toByteBuffer() for STRING type:
+ // Strings are encoded as UTF-8 bytes without length prefix
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] bytes = BinaryString.encodeUTF8(value.toString());
+ ensureCapacity(bytes.length);
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ void writeBytes(byte[] bytes) {
+ // Based on Iceberg's Conversions.toByteBuffer() for BINARY type:
+ // Binary data is stored directly without length prefix
+ ensureCapacity(bytes.length);
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ public void writeDecimal(Decimal value, int precision) {
+ // Iceberg stores decimals as unscaled big-endian byte arrays
+ // Reference:https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ byte[] unscaledBytes = value.toUnscaledBytes();
+
+ // For Iceberg, decimals are stored with fixed byte lengths based on precision
+ int requiredBytes = getIcebergDecimalBytes(precision);
+
+ byte[] icebergBytes = new byte[requiredBytes];
+
+ // Convert to big-endian format with proper padding
+ if (unscaledBytes.length <= requiredBytes) {
+ // Pad with sign extension
+ byte paddingByte =
+ (unscaledBytes.length > 0 && (unscaledBytes[0] & 0x80) != 0)
+ ? (byte) 0xFF
+ : (byte) 0x00;
+
+ Arrays.fill(icebergBytes, 0, requiredBytes - unscaledBytes.length, paddingByte);
+ System.arraycopy(
+ unscaledBytes,
+ 0,
+ icebergBytes,
+ requiredBytes - unscaledBytes.length,
+ unscaledBytes.length);
+ } else {
+ // Truncate if too large (should not happen with proper validation)
+ System.arraycopy(unscaledBytes, 0, icebergBytes, 0, requiredBytes);
+ }
+
+ writeBytes(icebergBytes);
+ }
+
+ public void writeTimestampNtz(TimestampNtz value, int precision) {
+ // Iceberg stores timestamps as microseconds since epoch
+ // Reference:
+ // https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ long micros = value.getMillisecond() * 1000L + (value.getNanoOfMillisecond() / 1000L);
+ writeLong(micros);
+ }
+
+ public void writeTimestampLtz(TimestampLtz value, int precision) {
+ // Iceberg stores timestamptz as microseconds since epoch in UTC
+ // Reference:
+ // https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ long epochMillis = value.getEpochMillisecond();
+ int nanoOfMilli = value.getNanoOfMillisecond();
+ long totalMicros = epochMillis * 1000L + (nanoOfMilli / 1000L);
+ writeLong(totalMicros);
+ }
+
+ private void ensureCapacity(int neededSize) {
+ if (buffer.length < cursor + neededSize) {
+ grow(cursor + neededSize);
+ }
+ }
+
+ private void grow(int minCapacity) {
+ int oldCapacity = buffer.length;
+ int newCapacity = oldCapacity + (oldCapacity >> 1); // 1.5x growth
+ if (newCapacity < minCapacity) {
+ newCapacity = minCapacity;
+ }
+ setBuffer(Arrays.copyOf(buffer, newCapacity));
+ }
+
+ private void setBuffer(byte[] buffer) {
+ this.buffer = buffer;
+ this.segment = MemorySegment.wrap(buffer);
+ }
+
+ /**
+ * Returns the number of bytes required to store a decimal with the given precision. Based on
+ * Iceberg's decimal storage specification.
+ *
+ * @param precision the decimal precision
+ * @return number of bytes required
+ */
+ private static int getIcebergDecimalBytes(int precision) {
+ // Reference:
+ // https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ if (precision <= 9) {
+ return 4; // Can fit in 4 bytes
+ } else if (precision <= 18) {
+ return 8; // Can fit in 8 bytes
+ } else {
+ return 16; // Requires 16 bytes
+ }
+ }
+
+ public ByteBuffer toByteBuffer() {
Review Comment:
removed
--
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] Iceberg encoding strategy [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2220919329
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link com.alibaba.fluss.row.InternalRow} using Iceberg's binary
+ * encoding format.
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal types use BIG-ENDIAN byte order
+ * <li>Strings are encoded as UTF-8 bytes
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ this.arity = arity;
+ // Conservative initial size to avoid frequent resizing
+ int initialSize = 8 + (arity * 8);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear only the used portion for efficiency
+ if (cursor > 0) {
+ Arrays.fill(buffer, 0, Math.min(cursor, buffer.length), (byte) 0);
+ }
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // For Iceberg key encoding, null values should not occur
+ // This is validated at the encoder level
+ throw new UnsupportedOperationException(
+ "Null values are not supported in Iceberg key encoding");
+ }
+
+ // todo: Revisit to add support as per Iceberg 1.9.1 post #1195 merge for Java 11 support
Review Comment:
Also, please create a issue to track 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/iceberg] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2221364032
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link com.alibaba.fluss.row.InternalRow} using Iceberg's binary
+ * encoding format.
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal types use BIG-ENDIAN byte order
+ * <li>Strings are encoded as UTF-8 bytes
+ * <li>Timestamps are stored as long values (microseconds since epoch)
+ * </ul>
+ *
+ * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
+ * but maintains byte-level compatibility with Iceberg's encoding.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ this.arity = arity;
+ // Conservative initial size to avoid frequent resizing
+ int initialSize = 8 + (arity * 8);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear only the used portion for efficiency
+ if (cursor > 0) {
+ Arrays.fill(buffer, 0, Math.min(cursor, buffer.length), (byte) 0);
+ }
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // For Iceberg key encoding, null values should not occur
+ // This is validated at the encoder level
+ throw new UnsupportedOperationException(
+ "Null values are not supported in Iceberg key encoding");
+ }
+
+ // todo: Revisit to add support as per Iceberg 1.9.1 post #1195 merge for Java 11 support
Review Comment:
https://github.com/apache/fluss/issues/1379
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on PR #1350:
URL: https://github.com/apache/fluss/pull/1350#issuecomment-3102675865
Thank you @luoyuxia for the review & feedbacks!
--
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] Iceberg encoding strategy [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2214684585
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.record.ChangeType;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/* This file is based on source code of Apache Iceberg Project (https://iceberg.apache.org/), licensed by the Apache
Review Comment:
Does it based on apache iceberg code? If not,
We can remove this since this file is not based on `Apache Iceberg Project`
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/KeyEncoder.java:
##########
@@ -46,6 +47,8 @@ static KeyEncoder of(
return CompactedKeyEncoder.createKeyEncoder(rowType, keyFields);
} else if (lakeFormat == DataLakeFormat.PAIMON) {
return new PaimonKeyEncoder(rowType, keyFields);
+ } else if (lakeFormat == DataLakeFormat.ICEBERG) {
+ return new IcebergKeyEncoder(rowType, keyFields);
Review Comment:
As the FIP said, we only support one key field currently.
```
checkArgument(
keyFields.size() == 1,
"key fields must have exactly one field for iceberg format.");
```
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.record.ChangeType;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/* This file is based on source code of Apache Iceberg Project (https://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding way.
+ *
+ * <p>The logic follows Iceberg's binary encoding format for row data using MemorySegment.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ // Verify MemorySegment uses little-endian for Iceberg compatibility
+ if (!MemorySegment.LITTLE_ENDIAN) {
Review Comment:
You can always use UnsafeUtils to put int, long, etc, since it will always be little endian.
```
UnsafeUtils.putLong(buffer, cursor, value);
```
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.record.ChangeType;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/* This file is based on source code of Apache Iceberg Project (https://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding way.
+ *
+ * <p>The logic follows Iceberg's binary encoding format for row data using MemorySegment.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ // Verify MemorySegment uses little-endian for Iceberg compatibility
+ if (!MemorySegment.LITTLE_ENDIAN) {
+ throw new IllegalStateException(
+ "MemorySegment must use little-endian byte order for Iceberg compatibility");
+ }
+
+ this.arity = arity;
+ // Initial buffer size estimation
+ int initialSize = calculateInitialSize(arity);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear buffer
+ Arrays.fill(buffer, 0, Math.min(buffer.length, calculateInitialSize(arity)), (byte) 0);
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // Iceberg handles nulls with a null marker byte
+ ensureCapacity(1);
+ segment.put(cursor, (byte) 0x00); // Null marker
+ cursor += 1;
+ }
+
+ public void writeChangeType(ChangeType kind) {
+ // TODO DISCUSS WITH YUXIA
+ // Iceberg doesn't store ChangeType in binary format like Paimon
+ // This is handled at the metadata/manifest level
+ // We skip this for Iceberg binary encoding
+ }
+
+ public void writeBoolean(int pos, boolean value) {
+ ensureCapacity(1);
+ segment.put(cursor, value ? (byte) 1 : (byte) 0);
+ cursor += 1;
+ }
+
+ public void writeByte(int pos, byte value) {
+ ensureCapacity(1);
+ segment.put(cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(int pos, short value) {
+ ensureCapacity(2);
+ segment.putShort(cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int pos, int value) {
+ ensureCapacity(4);
+ segment.putInt(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(int pos, long value) {
+ ensureCapacity(8);
+ segment.putLong(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(int pos, float value) {
+ ensureCapacity(4);
+ segment.putFloat(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(int pos, double value) {
+ ensureCapacity(8);
+ segment.putDouble(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(int pos, BinaryString input) {
+ if (input == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ byte[] bytes;
+ if (input.getSegments() == null) {
+ bytes = input.toString().getBytes(StandardCharsets.UTF_8);
+ } else {
+ int len = input.getSizeInBytes();
+ bytes = new byte[len];
+ extractFromFlussSegments(input, bytes);
+ }
+ writeBytes(pos, bytes);
+ }
+
+ void writeBytes(int pos, byte[] bytes) {
+ if (bytes == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg variable-length encoding: length prefix + data
+ ensureCapacity(4 + bytes.length);
+ segment.putInt(cursor, bytes.length);
+ cursor += 4;
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ public void writeDecimal(int pos, Decimal value, int precision) {
+ if (value == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg stores decimals as big-endian byte arrays
+ byte[] unscaledBytes = value.toUnscaledBytes();
+
+ // Calculate required bytes based on precision (following Iceberg's rules)
+ int requiredBytes = getIcebergDecimalBytes(precision);
+ ensureCapacity(4 + requiredBytes); // length + data
+
+ // Write length
+ segment.putInt(cursor, requiredBytes);
+ cursor += 4;
+
+ // Pad or truncate to required size
+ if (unscaledBytes.length <= requiredBytes) {
+ // Pad with sign extension
+ byte paddingByte =
+ (unscaledBytes.length > 0 && (unscaledBytes[0] & 0x80) != 0)
+ ? (byte) 0xFF
+ : (byte) 0x00;
+
+ // Write padding
+ for (int i = 0; i < requiredBytes - unscaledBytes.length; i++) {
+ segment.put(cursor + i, paddingByte);
+ }
+ // Write actual bytes
+ segment.put(
+ cursor + (requiredBytes - unscaledBytes.length),
+ unscaledBytes,
+ 0,
+ unscaledBytes.length);
+ } else {
+ // Truncate if too large
+ segment.put(cursor, unscaledBytes, 0, requiredBytes);
+ }
+ cursor += requiredBytes;
+ }
+
+ public void writeTimestampNtz(int pos, TimestampNtz value, int precision) {
+ if (value == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg stores timestamps as microseconds since epoch
+ ensureCapacity(8);
+ long micros = value.getMillisecond() * 1000L + (value.getNanoOfMillisecond() / 1000L);
+ segment.putLong(cursor, micros);
+ cursor += 8;
+ }
+
+ public void writeTimestampLtz(int pos, TimestampLtz value, int precision) {
+ if (value == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg stores timestamptz as microseconds since epoch in UTC
+ ensureCapacity(8);
+ long epochMillis = value.getEpochMillisecond();
+ int nanoOfMilli = value.getNanoOfMillisecond();
+ long totalMicros = epochMillis * 1000L + (nanoOfMilli / 1000L);
+ segment.putLong(cursor, totalMicros);
+ cursor += 8;
+ }
+
+ private void ensureCapacity(int neededSize) {
+ if (segment.size() < cursor + neededSize) {
+ grow(cursor + neededSize);
+ }
+ }
+
+ private void grow(int minCapacity) {
+ int oldCapacity = segment.size();
+ int newCapacity = Math.max(oldCapacity * 2, minCapacity);
+ setBuffer(Arrays.copyOf(buffer, newCapacity));
+ }
+
+ private void setBuffer(byte[] buffer) {
+ this.buffer = buffer;
+ this.segment = MemorySegment.wrap(buffer);
+ }
+
+ private void extractFromFlussSegments(BinaryString input, byte[] target) {
+ // Convert Fluss MemorySegments to byte array
+ MemorySegment[] segments = input.getSegments();
+ int offset = input.getOffset();
+ int length = input.getSizeInBytes();
+
+ if (segments.length == 1) {
+ segments[0].get(offset, target, 0, length);
+ } else {
+ copyFromFlussSegments(segments, offset, target, 0, length);
+ }
+ }
+
+ private void copyFromFlussSegments(
+ MemorySegment[] segments, int offset, byte[] target, int targetOffset, int length) {
+ int remaining = length;
+ int currentOffset = offset;
+ int currentTargetOffset = targetOffset;
+
+ for (MemorySegment segment : segments) {
+ if (remaining <= 0) {
+ break;
+ }
+
+ int segmentSize = segment.size();
+ if (currentOffset >= segmentSize) {
+ currentOffset -= segmentSize;
+ continue;
+ }
+
+ int availableInSegment = segmentSize - currentOffset;
+ int toCopy = Math.min(remaining, availableInSegment);
+
+ // Copy from MemorySegment to byte array
+ segment.get(currentOffset, target, currentTargetOffset, toCopy);
+
+ remaining -= toCopy;
+ currentTargetOffset += toCopy;
+ currentOffset = 0;
+ }
+ }
+
+ private static int calculateInitialSize(int arity) {
+ // Conservative estimate for Iceberg format
+ return 64 + (arity * 16); // Allow for variable length fields
Review Comment:
Why use `64 + (arity * 16);`? I'd suggest to use `8 + (arity * 8);` for space save. No need to allocate a larger byte array.
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.record.ChangeType;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/* This file is based on source code of Apache Iceberg Project (https://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding way.
+ *
+ * <p>The logic follows Iceberg's binary encoding format for row data using MemorySegment.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ // Verify MemorySegment uses little-endian for Iceberg compatibility
+ if (!MemorySegment.LITTLE_ENDIAN) {
+ throw new IllegalStateException(
+ "MemorySegment must use little-endian byte order for Iceberg compatibility");
+ }
+
+ this.arity = arity;
+ // Initial buffer size estimation
+ int initialSize = calculateInitialSize(arity);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear buffer
+ Arrays.fill(buffer, 0, Math.min(buffer.length, calculateInitialSize(arity)), (byte) 0);
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // Iceberg handles nulls with a null marker byte
+ ensureCapacity(1);
+ segment.put(cursor, (byte) 0x00); // Null marker
+ cursor += 1;
+ }
+
+ public void writeChangeType(ChangeType kind) {
+ // TODO DISCUSS WITH YUXIA
+ // Iceberg doesn't store ChangeType in binary format like Paimon
+ // This is handled at the metadata/manifest level
+ // We skip this for Iceberg binary encoding
+ }
+
+ public void writeBoolean(int pos, boolean value) {
+ ensureCapacity(1);
+ segment.put(cursor, value ? (byte) 1 : (byte) 0);
+ cursor += 1;
+ }
+
+ public void writeByte(int pos, byte value) {
+ ensureCapacity(1);
+ segment.put(cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(int pos, short value) {
+ ensureCapacity(2);
+ segment.putShort(cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int pos, int value) {
+ ensureCapacity(4);
+ segment.putInt(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(int pos, long value) {
+ ensureCapacity(8);
+ segment.putLong(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(int pos, float value) {
+ ensureCapacity(4);
+ segment.putFloat(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(int pos, double value) {
+ ensureCapacity(8);
+ segment.putDouble(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(int pos, BinaryString input) {
+ if (input == null) {
Review Comment:
For write string, let's reuse what we do in `CompactedKeyWriter` intead of introducing new piece of code:
```
public void writeString(int pos, BinaryString input) {
if (input.getSegments() == null) {
byte[] bytes = BinaryString.encodeUTF8(input.toString());
writeBytes(pos, bytes);
} else {
writeSegments(input.getSegments(), input.getOffset(), input.getSizeInBytes());
}
}
private void writeSegments(MemorySegment[] segments, int off, int len) {
writeInt(len);
if (len + off <= segments[0].size()) {
write(segments[0], off, len);
} else {
write(segments, off, len);
}
}
private void write(MemorySegment segment, int off, int len) {
ensureCapacity(len);
segment.get(off, this.buffer, this.cursor, len);
this.cursor += len;
}
private void write(MemorySegment[] segments, int off, int len) {
ensureCapacity(len);
int toWrite = len;
int fromOffset = off;
int toOffset = this.cursor;
for (MemorySegment sourceSegment : segments) {
int remain = sourceSegment.size() - fromOffset;
if (remain > 0) {
int localToWrite = Math.min(remain, toWrite);
sourceSegment.get(fromOffset, buffer, toOffset, localToWrite);
toWrite -= localToWrite;
toOffset += localToWrite;
fromOffset = 0;
} else {
fromOffset -= sourceSegment.size();
}
}
this.cursor += len;
}
```
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergKeyEncoder.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.encode.KeyEncoder;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.types.RowType;
+
+import java.util.List;
+
+/** An implementation of {@link KeyEncoder} to follow Iceberg's encoding strategy. */
+public class IcebergKeyEncoder implements KeyEncoder {
+
+ private final InternalRow.FieldGetter[] fieldGetters;
+
+ private final IcebergBinaryRowWriter.FieldWriter[] fieldEncoders;
+
+ private final IcebergBinaryRowWriter icebergBinaryRowWriter;
+
+ public IcebergKeyEncoder(RowType rowType, List<String> keys) {
+ // for get fields from fluss internal row
+ fieldGetters = new InternalRow.FieldGetter[keys.size()];
+ // for encode fields into iceberg
+ fieldEncoders = new IcebergBinaryRowWriter.FieldWriter[keys.size()];
+ for (int i = 0; i < keys.size(); i++) {
+ int keyIndex = rowType.getFieldIndex(keys.get(i));
+ DataType keyDataType = rowType.getTypeAt(keyIndex);
+ fieldGetters[i] = InternalRow.createFieldGetter(keyDataType, keyIndex);
+ fieldEncoders[i] = IcebergBinaryRowWriter.createFieldWriter(keyDataType);
+ }
+
+ icebergBinaryRowWriter = new IcebergBinaryRowWriter(keys.size());
+ }
+
+ @Override
+ public byte[] encodeKey(InternalRow row) {
+ icebergBinaryRowWriter.reset();
+
+ // TODO: Discuss with Yuxio Iceberg doesn't store ChangeType in binary format like Paimon
Review Comment:
we don't need store ChangeType, paimon store it just because we need to align to paimon's encoding since paimon always ecnoding change type. But icberg don't require it.
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.record.ChangeType;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/* This file is based on source code of Apache Iceberg Project (https://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding way.
Review Comment:
Add a link to refer to the code that iceberg encodes in iceberg repo for reference.
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.record.ChangeType;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/* This file is based on source code of Apache Iceberg Project (https://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding way.
+ *
+ * <p>The logic follows Iceberg's binary encoding format for row data using MemorySegment.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ // Verify MemorySegment uses little-endian for Iceberg compatibility
+ if (!MemorySegment.LITTLE_ENDIAN) {
+ throw new IllegalStateException(
+ "MemorySegment must use little-endian byte order for Iceberg compatibility");
+ }
+
+ this.arity = arity;
+ // Initial buffer size estimation
+ int initialSize = calculateInitialSize(arity);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear buffer
+ Arrays.fill(buffer, 0, Math.min(buffer.length, calculateInitialSize(arity)), (byte) 0);
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // Iceberg handles nulls with a null marker byte
+ ensureCapacity(1);
+ segment.put(cursor, (byte) 0x00); // Null marker
+ cursor += 1;
+ }
+
+ public void writeChangeType(ChangeType kind) {
Review Comment:
We don't need to write change type, we only use the key encoder to encode columns not change type.
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.record.ChangeType;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/* This file is based on source code of Apache Iceberg Project (https://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding way.
+ *
+ * <p>The logic follows Iceberg's binary encoding format for row data using MemorySegment.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ // Verify MemorySegment uses little-endian for Iceberg compatibility
+ if (!MemorySegment.LITTLE_ENDIAN) {
+ throw new IllegalStateException(
+ "MemorySegment must use little-endian byte order for Iceberg compatibility");
+ }
+
+ this.arity = arity;
+ // Initial buffer size estimation
+ int initialSize = calculateInitialSize(arity);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear buffer
+ Arrays.fill(buffer, 0, Math.min(buffer.length, calculateInitialSize(arity)), (byte) 0);
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // Iceberg handles nulls with a null marker byte
+ ensureCapacity(1);
+ segment.put(cursor, (byte) 0x00); // Null marker
+ cursor += 1;
+ }
+
+ public void writeChangeType(ChangeType kind) {
+ // TODO DISCUSS WITH YUXIA
+ // Iceberg doesn't store ChangeType in binary format like Paimon
+ // This is handled at the metadata/manifest level
+ // We skip this for Iceberg binary encoding
+ }
+
+ public void writeBoolean(int pos, boolean value) {
+ ensureCapacity(1);
+ segment.put(cursor, value ? (byte) 1 : (byte) 0);
+ cursor += 1;
+ }
+
+ public void writeByte(int pos, byte value) {
+ ensureCapacity(1);
+ segment.put(cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(int pos, short value) {
+ ensureCapacity(2);
+ segment.putShort(cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int pos, int value) {
+ ensureCapacity(4);
+ segment.putInt(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(int pos, long value) {
+ ensureCapacity(8);
+ segment.putLong(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(int pos, float value) {
+ ensureCapacity(4);
+ segment.putFloat(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(int pos, double value) {
+ ensureCapacity(8);
+ segment.putDouble(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(int pos, BinaryString input) {
+ if (input == null) {
Review Comment:
don't need to check is null or not since we already check it in
```
if (value == null) {
writer.setNullAt(pos);
}
```
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.record.ChangeType;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/* This file is based on source code of Apache Iceberg Project (https://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding way.
+ *
+ * <p>The logic follows Iceberg's binary encoding format for row data using MemorySegment.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ // Verify MemorySegment uses little-endian for Iceberg compatibility
+ if (!MemorySegment.LITTLE_ENDIAN) {
+ throw new IllegalStateException(
+ "MemorySegment must use little-endian byte order for Iceberg compatibility");
+ }
+
+ this.arity = arity;
+ // Initial buffer size estimation
+ int initialSize = calculateInitialSize(arity);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear buffer
+ Arrays.fill(buffer, 0, Math.min(buffer.length, calculateInitialSize(arity)), (byte) 0);
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // Iceberg handles nulls with a null marker byte
+ ensureCapacity(1);
+ segment.put(cursor, (byte) 0x00); // Null marker
+ cursor += 1;
+ }
+
+ public void writeChangeType(ChangeType kind) {
+ // TODO DISCUSS WITH YUXIA
+ // Iceberg doesn't store ChangeType in binary format like Paimon
+ // This is handled at the metadata/manifest level
+ // We skip this for Iceberg binary encoding
+ }
+
+ public void writeBoolean(int pos, boolean value) {
+ ensureCapacity(1);
+ segment.put(cursor, value ? (byte) 1 : (byte) 0);
+ cursor += 1;
+ }
+
+ public void writeByte(int pos, byte value) {
+ ensureCapacity(1);
+ segment.put(cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(int pos, short value) {
+ ensureCapacity(2);
+ segment.putShort(cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int pos, int value) {
+ ensureCapacity(4);
+ segment.putInt(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(int pos, long value) {
+ ensureCapacity(8);
+ segment.putLong(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(int pos, float value) {
+ ensureCapacity(4);
+ segment.putFloat(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(int pos, double value) {
+ ensureCapacity(8);
+ segment.putDouble(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(int pos, BinaryString input) {
+ if (input == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ byte[] bytes;
+ if (input.getSegments() == null) {
+ bytes = input.toString().getBytes(StandardCharsets.UTF_8);
+ } else {
+ int len = input.getSizeInBytes();
+ bytes = new byte[len];
+ extractFromFlussSegments(input, bytes);
+ }
+ writeBytes(pos, bytes);
+ }
+
+ void writeBytes(int pos, byte[] bytes) {
+ if (bytes == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg variable-length encoding: length prefix + data
+ ensureCapacity(4 + bytes.length);
+ segment.putInt(cursor, bytes.length);
+ cursor += 4;
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ public void writeDecimal(int pos, Decimal value, int precision) {
+ if (value == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg stores decimals as big-endian byte arrays
+ byte[] unscaledBytes = value.toUnscaledBytes();
+
+ // Calculate required bytes based on precision (following Iceberg's rules)
+ int requiredBytes = getIcebergDecimalBytes(precision);
+ ensureCapacity(4 + requiredBytes); // length + data
+
+ // Write length
+ segment.putInt(cursor, requiredBytes);
+ cursor += 4;
+
+ // Pad or truncate to required size
+ if (unscaledBytes.length <= requiredBytes) {
+ // Pad with sign extension
+ byte paddingByte =
+ (unscaledBytes.length > 0 && (unscaledBytes[0] & 0x80) != 0)
+ ? (byte) 0xFF
+ : (byte) 0x00;
+
+ // Write padding
+ for (int i = 0; i < requiredBytes - unscaledBytes.length; i++) {
+ segment.put(cursor + i, paddingByte);
+ }
+ // Write actual bytes
+ segment.put(
+ cursor + (requiredBytes - unscaledBytes.length),
+ unscaledBytes,
+ 0,
+ unscaledBytes.length);
+ } else {
+ // Truncate if too large
+ segment.put(cursor, unscaledBytes, 0, requiredBytes);
+ }
+ cursor += requiredBytes;
+ }
+
+ public void writeTimestampNtz(int pos, TimestampNtz value, int precision) {
+ if (value == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg stores timestamps as microseconds since epoch
+ ensureCapacity(8);
+ long micros = value.getMillisecond() * 1000L + (value.getNanoOfMillisecond() / 1000L);
+ segment.putLong(cursor, micros);
+ cursor += 8;
+ }
+
+ public void writeTimestampLtz(int pos, TimestampLtz value, int precision) {
+ if (value == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg stores timestamptz as microseconds since epoch in UTC
+ ensureCapacity(8);
+ long epochMillis = value.getEpochMillisecond();
+ int nanoOfMilli = value.getNanoOfMillisecond();
+ long totalMicros = epochMillis * 1000L + (nanoOfMilli / 1000L);
+ segment.putLong(cursor, totalMicros);
+ cursor += 8;
+ }
+
+ private void ensureCapacity(int neededSize) {
+ if (segment.size() < cursor + neededSize) {
+ grow(cursor + neededSize);
+ }
+ }
+
+ private void grow(int minCapacity) {
+ int oldCapacity = segment.size();
+ int newCapacity = Math.max(oldCapacity * 2, minCapacity);
+ setBuffer(Arrays.copyOf(buffer, newCapacity));
+ }
+
+ private void setBuffer(byte[] buffer) {
+ this.buffer = buffer;
+ this.segment = MemorySegment.wrap(buffer);
+ }
+
+ private void extractFromFlussSegments(BinaryString input, byte[] target) {
+ // Convert Fluss MemorySegments to byte array
+ MemorySegment[] segments = input.getSegments();
+ int offset = input.getOffset();
+ int length = input.getSizeInBytes();
+
+ if (segments.length == 1) {
+ segments[0].get(offset, target, 0, length);
+ } else {
+ copyFromFlussSegments(segments, offset, target, 0, length);
+ }
+ }
+
+ private void copyFromFlussSegments(
+ MemorySegment[] segments, int offset, byte[] target, int targetOffset, int length) {
+ int remaining = length;
+ int currentOffset = offset;
+ int currentTargetOffset = targetOffset;
+
+ for (MemorySegment segment : segments) {
+ if (remaining <= 0) {
+ break;
+ }
+
+ int segmentSize = segment.size();
+ if (currentOffset >= segmentSize) {
+ currentOffset -= segmentSize;
+ continue;
+ }
+
+ int availableInSegment = segmentSize - currentOffset;
+ int toCopy = Math.min(remaining, availableInSegment);
+
+ // Copy from MemorySegment to byte array
+ segment.get(currentOffset, target, currentTargetOffset, toCopy);
+
+ remaining -= toCopy;
+ currentTargetOffset += toCopy;
+ currentOffset = 0;
+ }
+ }
+
+ private static int calculateInitialSize(int arity) {
+ // Conservative estimate for Iceberg format
+ return 64 + (arity * 16); // Allow for variable length fields
+ }
+
+ private static int getIcebergDecimalBytes(int precision) {
+ // Iceberg's decimal storage requirements
+ if (precision <= 9) {
+ return 4;
+ }
+ if (precision <= 18) {
+ return 8;
+ }
+ return 16;
+ }
+
+ /**
+ * Creates an accessor for writing the elements of an iceberg binary row writer during runtime.
+ *
+ * @param fieldType the field type of the indexed row
+ */
+ public static FieldWriter createFieldWriter(DataType fieldType) {
+ final FieldWriter fieldWriter;
+ switch (fieldType.getTypeRoot()) {
+ case CHAR:
+ case STRING:
+ fieldWriter = (writer, pos, value) -> writer.writeString(pos, (BinaryString) value);
Review Comment:
nit: simplify to
```
writer.writeString((BinaryString) value);
```
The other writers can also be simplified.
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.record.ChangeType;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/* This file is based on source code of Apache Iceberg Project (https://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding way.
+ *
+ * <p>The logic follows Iceberg's binary encoding format for row data using MemorySegment.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ // Verify MemorySegment uses little-endian for Iceberg compatibility
+ if (!MemorySegment.LITTLE_ENDIAN) {
+ throw new IllegalStateException(
+ "MemorySegment must use little-endian byte order for Iceberg compatibility");
+ }
+
+ this.arity = arity;
+ // Initial buffer size estimation
+ int initialSize = calculateInitialSize(arity);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear buffer
+ Arrays.fill(buffer, 0, Math.min(buffer.length, calculateInitialSize(arity)), (byte) 0);
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // Iceberg handles nulls with a null marker byte
+ ensureCapacity(1);
+ segment.put(cursor, (byte) 0x00); // Null marker
+ cursor += 1;
+ }
+
+ public void writeChangeType(ChangeType kind) {
+ // TODO DISCUSS WITH YUXIA
+ // Iceberg doesn't store ChangeType in binary format like Paimon
+ // This is handled at the metadata/manifest level
+ // We skip this for Iceberg binary encoding
+ }
+
+ public void writeBoolean(int pos, boolean value) {
+ ensureCapacity(1);
+ segment.put(cursor, value ? (byte) 1 : (byte) 0);
+ cursor += 1;
+ }
+
+ public void writeByte(int pos, byte value) {
+ ensureCapacity(1);
+ segment.put(cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(int pos, short value) {
+ ensureCapacity(2);
+ segment.putShort(cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int pos, int value) {
+ ensureCapacity(4);
+ segment.putInt(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(int pos, long value) {
+ ensureCapacity(8);
+ segment.putLong(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(int pos, float value) {
+ ensureCapacity(4);
+ segment.putFloat(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(int pos, double value) {
+ ensureCapacity(8);
+ segment.putDouble(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(int pos, BinaryString input) {
+ if (input == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ byte[] bytes;
+ if (input.getSegments() == null) {
+ bytes = input.toString().getBytes(StandardCharsets.UTF_8);
+ } else {
+ int len = input.getSizeInBytes();
+ bytes = new byte[len];
+ extractFromFlussSegments(input, bytes);
+ }
+ writeBytes(pos, bytes);
+ }
+
+ void writeBytes(int pos, byte[] bytes) {
+ if (bytes == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg variable-length encoding: length prefix + data
+ ensureCapacity(4 + bytes.length);
+ segment.putInt(cursor, bytes.length);
+ cursor += 4;
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ public void writeDecimal(int pos, Decimal value, int precision) {
+ if (value == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg stores decimals as big-endian byte arrays
+ byte[] unscaledBytes = value.toUnscaledBytes();
+
+ // Calculate required bytes based on precision (following Iceberg's rules)
+ int requiredBytes = getIcebergDecimalBytes(precision);
Review Comment:
Where can i get the logic in iceberg repo?
##########
fluss-common/src/test/java/com/alibaba/fluss/row/encode/iceberg/IcebergKeyEncoderTest.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.row.indexed.IndexedRow;
+import com.alibaba.fluss.row.indexed.IndexedRowWriter;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.types.DataTypes;
+import com.alibaba.fluss.types.RowType;
+import com.alibaba.fluss.utils.TypeUtils;
+
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.List;
+
+import static com.alibaba.fluss.row.TestInternalRowGenerator.createAllRowType;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** UT for {@link IcebergKeyEncoder} to verify the encoding result matches Iceberg's format. */
+class IcebergKeyEncoderTest {
+
+ @Test
+ void testEncodeKey() {
+ // create a row with all types
+ RowType allRowType = createAllRowType();
+ DataType[] allDataTypes = allRowType.getChildren().toArray(new DataType[0]);
+
+ IndexedRow indexedRow = genFlussRowForAllTypes(allDataTypes);
+ List<String> encodedKeys = allRowType.getFieldNames();
+ IcebergKeyEncoder icebergKeyEncoder = new IcebergKeyEncoder(allRowType, encodedKeys);
+
+ // encode with Fluss own implementation for Iceberg
+ byte[] encodedKey = icebergKeyEncoder.encodeKey(indexedRow);
+
+ // encode with Iceberg implementation using native Iceberg classes
Review Comment:
This is still not using native Iceberg classes to encode the keys. It still use our own IcebergBinaryRowWriter, so that we can't ensure our encoding is align with iceberg.
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.record.ChangeType;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/* This file is based on source code of Apache Iceberg Project (https://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding way.
+ *
+ * <p>The logic follows Iceberg's binary encoding format for row data using MemorySegment.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ // Verify MemorySegment uses little-endian for Iceberg compatibility
+ if (!MemorySegment.LITTLE_ENDIAN) {
+ throw new IllegalStateException(
+ "MemorySegment must use little-endian byte order for Iceberg compatibility");
+ }
+
+ this.arity = arity;
+ // Initial buffer size estimation
+ int initialSize = calculateInitialSize(arity);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear buffer
+ Arrays.fill(buffer, 0, Math.min(buffer.length, calculateInitialSize(arity)), (byte) 0);
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // Iceberg handles nulls with a null marker byte
+ ensureCapacity(1);
+ segment.put(cursor, (byte) 0x00); // Null marker
+ cursor += 1;
+ }
+
+ public void writeChangeType(ChangeType kind) {
+ // TODO DISCUSS WITH YUXIA
+ // Iceberg doesn't store ChangeType in binary format like Paimon
+ // This is handled at the metadata/manifest level
+ // We skip this for Iceberg binary encoding
+ }
+
+ public void writeBoolean(int pos, boolean value) {
+ ensureCapacity(1);
+ segment.put(cursor, value ? (byte) 1 : (byte) 0);
+ cursor += 1;
+ }
+
+ public void writeByte(int pos, byte value) {
+ ensureCapacity(1);
+ segment.put(cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(int pos, short value) {
+ ensureCapacity(2);
+ segment.putShort(cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int pos, int value) {
+ ensureCapacity(4);
+ segment.putInt(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(int pos, long value) {
+ ensureCapacity(8);
+ segment.putLong(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(int pos, float value) {
+ ensureCapacity(4);
+ segment.putFloat(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(int pos, double value) {
+ ensureCapacity(8);
+ segment.putDouble(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(int pos, BinaryString input) {
+ if (input == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ byte[] bytes;
+ if (input.getSegments() == null) {
+ bytes = input.toString().getBytes(StandardCharsets.UTF_8);
+ } else {
+ int len = input.getSizeInBytes();
+ bytes = new byte[len];
+ extractFromFlussSegments(input, bytes);
+ }
+ writeBytes(pos, bytes);
+ }
+
+ void writeBytes(int pos, byte[] bytes) {
+ if (bytes == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg variable-length encoding: length prefix + data
+ ensureCapacity(4 + bytes.length);
+ segment.putInt(cursor, bytes.length);
+ cursor += 4;
+ segment.put(cursor, bytes, 0, bytes.length);
+ cursor += bytes.length;
+ }
+
+ public void writeDecimal(int pos, Decimal value, int precision) {
+ if (value == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg stores decimals as big-endian byte arrays
+ byte[] unscaledBytes = value.toUnscaledBytes();
+
+ // Calculate required bytes based on precision (following Iceberg's rules)
+ int requiredBytes = getIcebergDecimalBytes(precision);
+ ensureCapacity(4 + requiredBytes); // length + data
+
+ // Write length
+ segment.putInt(cursor, requiredBytes);
+ cursor += 4;
+
+ // Pad or truncate to required size
+ if (unscaledBytes.length <= requiredBytes) {
+ // Pad with sign extension
+ byte paddingByte =
+ (unscaledBytes.length > 0 && (unscaledBytes[0] & 0x80) != 0)
+ ? (byte) 0xFF
+ : (byte) 0x00;
+
+ // Write padding
+ for (int i = 0; i < requiredBytes - unscaledBytes.length; i++) {
+ segment.put(cursor + i, paddingByte);
+ }
+ // Write actual bytes
+ segment.put(
+ cursor + (requiredBytes - unscaledBytes.length),
+ unscaledBytes,
+ 0,
+ unscaledBytes.length);
+ } else {
+ // Truncate if too large
+ segment.put(cursor, unscaledBytes, 0, requiredBytes);
+ }
+ cursor += requiredBytes;
+ }
+
+ public void writeTimestampNtz(int pos, TimestampNtz value, int precision) {
+ if (value == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg stores timestamps as microseconds since epoch
+ ensureCapacity(8);
+ long micros = value.getMillisecond() * 1000L + (value.getNanoOfMillisecond() / 1000L);
+ segment.putLong(cursor, micros);
+ cursor += 8;
+ }
+
+ public void writeTimestampLtz(int pos, TimestampLtz value, int precision) {
+ if (value == null) {
+ setNullAt(pos);
+ return;
+ }
+
+ // Iceberg stores timestamptz as microseconds since epoch in UTC
+ ensureCapacity(8);
+ long epochMillis = value.getEpochMillisecond();
+ int nanoOfMilli = value.getNanoOfMillisecond();
+ long totalMicros = epochMillis * 1000L + (nanoOfMilli / 1000L);
+ segment.putLong(cursor, totalMicros);
+ cursor += 8;
+ }
+
+ private void ensureCapacity(int neededSize) {
+ if (segment.size() < cursor + neededSize) {
+ grow(cursor + neededSize);
+ }
+ }
+
+ private void grow(int minCapacity) {
+ int oldCapacity = segment.size();
+ int newCapacity = Math.max(oldCapacity * 2, minCapacity);
+ setBuffer(Arrays.copyOf(buffer, newCapacity));
+ }
+
+ private void setBuffer(byte[] buffer) {
+ this.buffer = buffer;
+ this.segment = MemorySegment.wrap(buffer);
+ }
+
+ private void extractFromFlussSegments(BinaryString input, byte[] target) {
+ // Convert Fluss MemorySegments to byte array
+ MemorySegment[] segments = input.getSegments();
+ int offset = input.getOffset();
+ int length = input.getSizeInBytes();
+
+ if (segments.length == 1) {
+ segments[0].get(offset, target, 0, length);
+ } else {
+ copyFromFlussSegments(segments, offset, target, 0, length);
+ }
+ }
+
+ private void copyFromFlussSegments(
+ MemorySegment[] segments, int offset, byte[] target, int targetOffset, int length) {
+ int remaining = length;
+ int currentOffset = offset;
+ int currentTargetOffset = targetOffset;
+
+ for (MemorySegment segment : segments) {
+ if (remaining <= 0) {
+ break;
+ }
+
+ int segmentSize = segment.size();
+ if (currentOffset >= segmentSize) {
+ currentOffset -= segmentSize;
+ continue;
+ }
+
+ int availableInSegment = segmentSize - currentOffset;
+ int toCopy = Math.min(remaining, availableInSegment);
+
+ // Copy from MemorySegment to byte array
+ segment.get(currentOffset, target, currentTargetOffset, toCopy);
+
+ remaining -= toCopy;
+ currentTargetOffset += toCopy;
+ currentOffset = 0;
+ }
+ }
+
+ private static int calculateInitialSize(int arity) {
+ // Conservative estimate for Iceberg format
+ return 64 + (arity * 16); // Allow for variable length fields
+ }
+
+ private static int getIcebergDecimalBytes(int precision) {
+ // Iceberg's decimal storage requirements
+ if (precision <= 9) {
+ return 4;
+ }
+ if (precision <= 18) {
+ return 8;
+ }
+ return 16;
+ }
+
+ /**
+ * Creates an accessor for writing the elements of an iceberg binary row writer during runtime.
+ *
+ * @param fieldType the field type of the indexed row
+ */
+ public static FieldWriter createFieldWriter(DataType fieldType) {
+ final FieldWriter fieldWriter;
+ switch (fieldType.getTypeRoot()) {
+ case CHAR:
+ case STRING:
+ fieldWriter = (writer, pos, value) -> writer.writeString(pos, (BinaryString) value);
+ break;
+ case BOOLEAN:
+ fieldWriter = (writer, pos, value) -> writer.writeBoolean(pos, (boolean) value);
+ break;
+ case BINARY:
+ case BYTES:
+ fieldWriter = (writer, pos, value) -> writer.writeBytes(pos, (byte[]) value);
+ break;
+ case DECIMAL:
+ final int decimalPrecision = getPrecision(fieldType);
+ fieldWriter =
+ (writer, pos, value) ->
+ writer.writeDecimal(pos, (Decimal) value, decimalPrecision);
+ break;
+ case TINYINT:
+ fieldWriter = (writer, pos, value) -> writer.writeByte(pos, (byte) value);
+ break;
+ case SMALLINT:
+ fieldWriter = (writer, pos, value) -> writer.writeShort(pos, (short) value);
+ break;
+ case INTEGER:
+ case DATE:
+ case TIME_WITHOUT_TIME_ZONE:
+ fieldWriter = (writer, pos, value) -> writer.writeInt(pos, (int) value);
+ break;
+ case BIGINT:
+ fieldWriter = (writer, pos, value) -> writer.writeLong(pos, (long) value);
+ break;
+ case FLOAT:
+ fieldWriter = (writer, pos, value) -> writer.writeFloat(pos, (float) value);
+ break;
+ case DOUBLE:
+ fieldWriter = (writer, pos, value) -> writer.writeDouble(pos, (double) value);
+ break;
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ final int timestampNtzPrecision = getPrecision(fieldType);
+ fieldWriter =
+ (writer, pos, value) ->
+ writer.writeTimestampNtz(
+ pos, (TimestampNtz) value, timestampNtzPrecision);
+ break;
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ final int timestampLtzPrecision = getPrecision(fieldType);
+ fieldWriter =
+ (writer, pos, value) ->
+ writer.writeTimestampLtz(
+ pos, (TimestampLtz) value, timestampLtzPrecision);
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported type for Iceberg binary row writer: " + fieldType);
+ }
+
+ if (!fieldType.isNullable()) {
+ return fieldWriter;
+ }
+
+ return (writer, pos, value) -> {
+ if (value == null) {
+ writer.setNullAt(pos);
+ } else {
+ fieldWriter.writeField(writer, pos, value);
+ }
+ };
+ }
+
+ /** Accessor for writing the elements of an iceberg binary row writer during runtime. */
+ interface FieldWriter extends Serializable {
+ void writeField(IcebergBinaryRowWriter writer, int pos, Object value);
Review Comment:
nit:
```suggestion
void writeField(IcebergBinaryRowWriter writer, Object value);
```
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2216870707
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.record.ChangeType;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/* This file is based on source code of Apache Iceberg Project (https://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding way.
+ *
+ * <p>The logic follows Iceberg's binary encoding format for row data using MemorySegment.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ // Verify MemorySegment uses little-endian for Iceberg compatibility
+ if (!MemorySegment.LITTLE_ENDIAN) {
+ throw new IllegalStateException(
+ "MemorySegment must use little-endian byte order for Iceberg compatibility");
+ }
+
+ this.arity = arity;
+ // Initial buffer size estimation
+ int initialSize = calculateInitialSize(arity);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear buffer
+ Arrays.fill(buffer, 0, Math.min(buffer.length, calculateInitialSize(arity)), (byte) 0);
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // Iceberg handles nulls with a null marker byte
+ ensureCapacity(1);
+ segment.put(cursor, (byte) 0x00); // Null marker
+ cursor += 1;
+ }
+
+ public void writeChangeType(ChangeType kind) {
Review Comment:
sure, will remove 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/iceberg] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2217452617
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.record.ChangeType;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/* This file is based on source code of Apache Iceberg Project (https://iceberg.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding way.
+ *
+ * <p>The logic follows Iceberg's binary encoding format for row data using MemorySegment.
+ */
+class IcebergBinaryRowWriter {
+
+ private final int arity;
+ private byte[] buffer;
+ private MemorySegment segment;
+ private int cursor;
+
+ public IcebergBinaryRowWriter(int arity) {
+ // Verify MemorySegment uses little-endian for Iceberg compatibility
+ if (!MemorySegment.LITTLE_ENDIAN) {
+ throw new IllegalStateException(
+ "MemorySegment must use little-endian byte order for Iceberg compatibility");
+ }
+
+ this.arity = arity;
+ // Initial buffer size estimation
+ int initialSize = calculateInitialSize(arity);
+ setBuffer(new byte[initialSize]);
+ reset();
+ }
+
+ public void reset() {
+ this.cursor = 0;
+ // Clear buffer
+ Arrays.fill(buffer, 0, Math.min(buffer.length, calculateInitialSize(arity)), (byte) 0);
+ }
+
+ public byte[] toBytes() {
+ byte[] result = new byte[cursor];
+ System.arraycopy(buffer, 0, result, 0, cursor);
+ return result;
+ }
+
+ public void setNullAt(int pos) {
+ // Iceberg handles nulls with a null marker byte
+ ensureCapacity(1);
+ segment.put(cursor, (byte) 0x00); // Null marker
+ cursor += 1;
+ }
+
+ public void writeChangeType(ChangeType kind) {
+ // TODO DISCUSS WITH YUXIA
+ // Iceberg doesn't store ChangeType in binary format like Paimon
+ // This is handled at the metadata/manifest level
+ // We skip this for Iceberg binary encoding
+ }
+
+ public void writeBoolean(int pos, boolean value) {
+ ensureCapacity(1);
+ segment.put(cursor, value ? (byte) 1 : (byte) 0);
+ cursor += 1;
+ }
+
+ public void writeByte(int pos, byte value) {
+ ensureCapacity(1);
+ segment.put(cursor, value);
+ cursor += 1;
+ }
+
+ public void writeShort(int pos, short value) {
+ ensureCapacity(2);
+ segment.putShort(cursor, value);
+ cursor += 2;
+ }
+
+ public void writeInt(int pos, int value) {
+ ensureCapacity(4);
+ segment.putInt(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeLong(int pos, long value) {
+ ensureCapacity(8);
+ segment.putLong(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeFloat(int pos, float value) {
+ ensureCapacity(4);
+ segment.putFloat(cursor, value);
+ cursor += 4;
+ }
+
+ public void writeDouble(int pos, double value) {
+ ensureCapacity(8);
+ segment.putDouble(cursor, value);
+ cursor += 8;
+ }
+
+ public void writeString(int pos, BinaryString input) {
+ if (input == null) {
Review Comment:
Sure!
--
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] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2221542999
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
Review Comment:
I missed it, I could imagine now it make sense if we have two columns and want to bucket them using them as a encoded string, it would look like this [0,0,0,3, C,A,T, 0,0,0,3, D,O,G] |len=3| |CAT|(utf-8 data) |len=3| |DOG|utf-8 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/iceberg] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2219513535
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergKeyEncoder.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.encode.KeyEncoder;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.types.RowType;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import static com.alibaba.fluss.utils.Preconditions.checkArgument;
+
+/** An implementation of {@link KeyEncoder} to follow Iceberg's encoding strategy. */
+public class IcebergKeyEncoder implements KeyEncoder {
+
+ private final InternalRow.FieldGetter[] fieldGetters;
+
+ private final IcebergBinaryRowWriter.FieldWriter[] fieldEncoders;
+
+ private final IcebergBinaryRowWriter icebergBinaryRowWriter;
+
+ public IcebergKeyEncoder(RowType rowType, List<String> keys) {
+
+ // Validate single key field requirement as per FIP
+ checkArgument(
+ keys.size() == 1,
+ "Key fields must have exactly one field for iceberg format, but got: %s",
+ keys.size());
+
+ // for get fields from fluss internal row
+ fieldGetters = new InternalRow.FieldGetter[keys.size()];
+ // for encode fields into iceberg
+ fieldEncoders = new IcebergBinaryRowWriter.FieldWriter[keys.size()];
+ for (int i = 0; i < keys.size(); i++) {
+ int keyIndex = rowType.getFieldIndex(keys.get(i));
+ DataType keyDataType = rowType.getTypeAt(keyIndex);
+ fieldGetters[i] = InternalRow.createFieldGetter(keyDataType, keyIndex);
+ fieldEncoders[i] = IcebergBinaryRowWriter.createFieldWriter(keyDataType);
+ }
+
+ icebergBinaryRowWriter = new IcebergBinaryRowWriter(keys.size());
+ }
+
+ @Override
+ public byte[] encodeKey(InternalRow row) {
+ icebergBinaryRowWriter.reset();
+ // iterate all the fields of the row, and encode each field
+ for (int i = 0; i < fieldGetters.length; i++) {
+ fieldEncoders[i].writeField(
+ icebergBinaryRowWriter, fieldGetters[i].getFieldOrNull(row));
+ }
+ return icebergBinaryRowWriter.toBytes();
+ }
+
+ /**
+ * Returns the encoded key as a ByteBuffer for Iceberg-compatible operations. This is useful
+ * when interfacing with Iceberg's hash functions which expect ByteBuffer.
+ */
+ public ByteBuffer encodeKeyAsBuffer(InternalRow row) {
Review Comment:
removed it, i just wrote it as an optional method to match the iceberg encode return type, but we don't need 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/iceberg] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2219513535
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergKeyEncoder.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.encode.KeyEncoder;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.types.RowType;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import static com.alibaba.fluss.utils.Preconditions.checkArgument;
+
+/** An implementation of {@link KeyEncoder} to follow Iceberg's encoding strategy. */
+public class IcebergKeyEncoder implements KeyEncoder {
+
+ private final InternalRow.FieldGetter[] fieldGetters;
+
+ private final IcebergBinaryRowWriter.FieldWriter[] fieldEncoders;
+
+ private final IcebergBinaryRowWriter icebergBinaryRowWriter;
+
+ public IcebergKeyEncoder(RowType rowType, List<String> keys) {
+
+ // Validate single key field requirement as per FIP
+ checkArgument(
+ keys.size() == 1,
+ "Key fields must have exactly one field for iceberg format, but got: %s",
+ keys.size());
+
+ // for get fields from fluss internal row
+ fieldGetters = new InternalRow.FieldGetter[keys.size()];
+ // for encode fields into iceberg
+ fieldEncoders = new IcebergBinaryRowWriter.FieldWriter[keys.size()];
+ for (int i = 0; i < keys.size(); i++) {
+ int keyIndex = rowType.getFieldIndex(keys.get(i));
+ DataType keyDataType = rowType.getTypeAt(keyIndex);
+ fieldGetters[i] = InternalRow.createFieldGetter(keyDataType, keyIndex);
+ fieldEncoders[i] = IcebergBinaryRowWriter.createFieldWriter(keyDataType);
+ }
+
+ icebergBinaryRowWriter = new IcebergBinaryRowWriter(keys.size());
+ }
+
+ @Override
+ public byte[] encodeKey(InternalRow row) {
+ icebergBinaryRowWriter.reset();
+ // iterate all the fields of the row, and encode each field
+ for (int i = 0; i < fieldGetters.length; i++) {
+ fieldEncoders[i].writeField(
+ icebergBinaryRowWriter, fieldGetters[i].getFieldOrNull(row));
+ }
+ return icebergBinaryRowWriter.toBytes();
+ }
+
+ /**
+ * Returns the encoded key as a ByteBuffer for Iceberg-compatible operations. This is useful
+ * when interfacing with Iceberg's hash functions which expect ByteBuffer.
+ */
+ public ByteBuffer encodeKeyAsBuffer(InternalRow row) {
Review Comment:
removed it put as an optional method to match the iceberg encode return type, but we don't need 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/iceberg] Iceberg encoding strategy [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #1350:
URL: https://github.com/apache/fluss/pull/1350#discussion_r2219673244
##########
fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * 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 com.alibaba.fluss.row.encode.iceberg;
+
+import com.alibaba.fluss.memory.MemorySegment;
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+import com.alibaba.fluss.types.DataType;
+import com.alibaba.fluss.utils.UnsafeUtils;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
+
+/**
+ * A writer to encode Fluss's {@link InternalRow} using Iceberg's binary encoding format.
+ *
+ * <p>This implementation follows Iceberg's binary encoding specification for partition and bucket
+ * keys. Reference: https://iceberg.apache.org/spec/#partition-transforms
+ *
+ * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
+ * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
+ *
+ * <p>Key encoding principles from Iceberg's Conversions class:
+ *
+ * <ul>
+ * <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ * <li>Decimal and UUID types use BIG-ENDIAN byte order
+ * <li>NO length prefix for any type - the buffer size determines the length
Review Comment:
can you explain a bit why we need length? why it is so important?
--
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