You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "singhpk234 (via GitHub)" <gi...@apache.org> on 2024/10/17 22:14:21 UTC
[PR] [KafkaConnect] Fix RecordConverter [iceberg]
singhpk234 opened a new pull request, #11346:
URL: https://github.com/apache/iceberg/pull/11346
## About the change
The UUID type in the parquet writer expects ByteBuffer rather than UUID otherwise writer fails with :
```
class java.util.UUID cannot be cast to class [B (java.util.UUID and [B are in module java.base of loader 'bootstrap')
```
The FixedLength would need byteArray rather than ByteBuffer otherwise one get this error
```
class java.nio.HeapByteBuffer cannot be cast to class [B (java.nio.HeapByteBuffer and [B are in module java.base of loader 'bootstrap')
```
will add the test cases shortly.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "ajantha-bhat (via GitHub)" <gi...@apache.org>.
ajantha-bhat commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1811920370
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java:
##########
@@ -206,11 +227,16 @@ public static void beforeAll() {
@BeforeEach
public void before() {
+ super.before();
+ ImmutableMap<String, String> mp = ImmutableMap.of("write.format.default", format);
+ System.out.println(format);
Review Comment:
please remove these sout
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1815718572
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java:
##########
@@ -84,11 +93,18 @@
import org.apache.kafka.connect.storage.ConverterType;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
-public class RecordConverterTest {
+@ExtendWith(ParameterizedTestExtension.class)
+public class RecordConverterTest extends BaseWriterTest {
Review Comment:
Shouldn't we have this parameterized by file type? The tests here make sense to me but I am only looking at this module for the first time for this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1813021695
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java:
##########
@@ -84,11 +93,18 @@
import org.apache.kafka.connect.storage.ConverterType;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
-public class RecordConverterTest {
+@ExtendWith(ParameterizedTestExtension.class)
+public class RecordConverterTest extends BaseWriterTest {
Review Comment:
Maybe we can create a dedicated test class for writer ?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1811346546
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java:
##########
@@ -84,11 +93,18 @@
import org.apache.kafka.connect.storage.ConverterType;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
-public class RecordConverterTest {
+@ExtendWith(ParameterizedTestExtension.class)
+public class RecordConverterTest extends BaseWriterTest {
Review Comment:
Was thinking in lines of conversion functions are no longer format agnostic as we are adding format info into deciding the record conversion, hence though it would be fair to test this E2E here.
please let me know your thoughts considering above.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1815721238
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java:
##########
@@ -921,11 +948,26 @@ private void assertRecordValues(Record record) {
assertThat(rec.getField("dec")).isEqualTo(DEC_VAL);
assertThat(rec.getField("s")).isEqualTo(STR_VAL);
assertThat(rec.getField("b")).isEqualTo(true);
- assertThat(rec.getField("u")).isEqualTo(UUID_VAL);
- assertThat(rec.getField("f")).isEqualTo(BYTES_VAL);
+ assertThat(rec.getField("f")).isEqualTo(BYTES_VAL.array());
assertThat(rec.getField("bi")).isEqualTo(BYTES_VAL);
assertThat(rec.getField("li")).isEqualTo(LIST_VAL);
assertThat(rec.getField("ma")).isEqualTo(MAP_VAL);
+
+ IcebergSinkConfig icebergSinkConfig = mock(IcebergSinkConfig.class);
+ when(icebergSinkConfig.tableConfig(any())).thenReturn(mock(TableSinkConfig.class));
+ when(icebergSinkConfig.writeProps())
+ .thenReturn(ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format));
+ WriteResult result =
+ writeTest(ImmutableList.of(rec), icebergSinkConfig, UnpartitionedWriter.class);
+
+ if (format.equals(FileFormat.PARQUET.name().toLowerCase(Locale.ROOT))) {
+ assertThat(rec.getField("u")).isEqualTo(UUIDUtil.convert(UUID_VAL));
Review Comment:
In generally I don't like replicating the implementation logic in the test. Can we check this in a more agnostic way?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "bryanck (via GitHub)" <gi...@apache.org>.
bryanck commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1811332848
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java:
##########
@@ -126,10 +127,14 @@ private Object convertValue(
case STRING:
return convertString(value);
case UUID:
+ if ("parquet".equals(config.writeProps().get("write.format.default"))) {
+ return UUIDUtil.convert(convertUUID(value));
Review Comment:
We could move the call to `UUIDUtil.convert()` into the `convertUUID()` method I think.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1806749659
##########
orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriters.java:
##########
@@ -101,8 +101,8 @@ public static OrcValueWriter<ByteBuffer> byteBuffers() {
return ByteBufferWriter.INSTANCE;
}
- public static OrcValueWriter<UUID> uuids() {
- return UUIDWriter.INSTANCE;
+ public static OrcValueWriter<ByteBuffer> uuids() {
+ return ByteBufferWriter.INSTANCE;
Review Comment:
removed this and handled this in the record converted
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer merged PR #11346:
URL: https://github.com/apache/iceberg/pull/11346
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1815720005
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java:
##########
@@ -84,11 +93,18 @@
import org.apache.kafka.connect.storage.ConverterType;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
-public class RecordConverterTest {
+@ExtendWith(ParameterizedTestExtension.class)
+public class RecordConverterTest extends BaseWriterTest {
Review Comment:
Ah iI see, the format only comes into play for UUID so the other parameterizations are essentially no-ops. Perhaps we just need one specalized test then ' testParquetUUIDSerialization"
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java:
##########
@@ -84,11 +93,18 @@
import org.apache.kafka.connect.storage.ConverterType;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
-public class RecordConverterTest {
+@ExtendWith(ParameterizedTestExtension.class)
+public class RecordConverterTest extends BaseWriterTest {
Review Comment:
Ah iI see, the format only comes into play for UUID so the other parameterizations are essentially no-ops. Perhaps we just need one specialized test then ' testParquetUUIDSerialization"
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "Gezi-lzq (via GitHub)" <gi...@apache.org>.
Gezi-lzq commented on PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#issuecomment-2505163794
When writing UUIDs, should we handle the conversion directly within BaseParquetWriter, by modifying BaseParquetWriter#primitive to check if the LogicalTypeAnnotation is a UUID and then use a UUIDWriter for writing it, instead of performing the conversion based on the file type before writing?
```java
@Override
public ParquetValueWriter<?> primitive(PrimitiveType primitive) {
....
switch (primitive.getPrimitiveTypeName()) {
case FIXED_LEN_BYTE_ARRAY:
if (LogicalTypeAnnotation.uuidType().equals(primitive.getLogicalTypeAnnotation())) {
return new UUIDWriter(desc);
}
return new FixedWriter(desc);
...
}
}
private static class UUIDWriter extends ParquetValueWriters.PrimitiveWriter<UUID> {
private UUIDWriter(ColumnDescriptor desc) {
super(desc);
}
@Override
public void write(int repetitionLevel, UUID value) {
column.writeBinary(repetitionLevel, Binary.fromReusedByteArray(UUIDUtil.convert(value)));
}
}
```
Similar to the approach taken in [Issue #4038 in Apache Iceberg](https://github.com/apache/iceberg/issues/4038).
@singhpk234 @bryanck @openinx
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1811926112
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java:
##########
@@ -206,11 +227,16 @@ public static void beforeAll() {
@BeforeEach
public void before() {
+ super.before();
+ ImmutableMap<String, String> mp = ImmutableMap.of("write.format.default", format);
+ System.out.println(format);
Review Comment:
Oops looks like I missed removing these lines, thanks for catching !
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1815762264
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java:
##########
@@ -921,11 +948,26 @@ private void assertRecordValues(Record record) {
assertThat(rec.getField("dec")).isEqualTo(DEC_VAL);
assertThat(rec.getField("s")).isEqualTo(STR_VAL);
assertThat(rec.getField("b")).isEqualTo(true);
- assertThat(rec.getField("u")).isEqualTo(UUID_VAL);
- assertThat(rec.getField("f")).isEqualTo(BYTES_VAL);
+ assertThat(rec.getField("f")).isEqualTo(BYTES_VAL.array());
assertThat(rec.getField("bi")).isEqualTo(BYTES_VAL);
assertThat(rec.getField("li")).isEqualTo(LIST_VAL);
assertThat(rec.getField("ma")).isEqualTo(MAP_VAL);
+
+ IcebergSinkConfig icebergSinkConfig = mock(IcebergSinkConfig.class);
+ when(icebergSinkConfig.tableConfig(any())).thenReturn(mock(TableSinkConfig.class));
+ when(icebergSinkConfig.writeProps())
+ .thenReturn(ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format));
+ WriteResult result =
+ writeTest(ImmutableList.of(rec), icebergSinkConfig, UnpartitionedWriter.class);
+
+ if (format.equals(FileFormat.PARQUET.name().toLowerCase(Locale.ROOT))) {
+ assertThat(rec.getField("u")).isEqualTo(UUIDUtil.convert(UUID_VAL));
Review Comment:
make sense ! Addressed the feedback.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1806695970
##########
orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriters.java:
##########
@@ -101,8 +101,8 @@ public static OrcValueWriter<ByteBuffer> byteBuffers() {
return ByteBufferWriter.INSTANCE;
}
- public static OrcValueWriter<UUID> uuids() {
- return UUIDWriter.INSTANCE;
+ public static OrcValueWriter<ByteBuffer> uuids() {
+ return ByteBufferWriter.INSTANCE;
Review Comment:
Need to revisit this.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java:
##########
@@ -921,11 +929,24 @@ private void assertRecordValues(Record record) {
assertThat(rec.getField("dec")).isEqualTo(DEC_VAL);
assertThat(rec.getField("s")).isEqualTo(STR_VAL);
assertThat(rec.getField("b")).isEqualTo(true);
- assertThat(rec.getField("u")).isEqualTo(UUID_VAL);
- assertThat(rec.getField("f")).isEqualTo(BYTES_VAL);
+ assertThat(rec.getField("u")).isEqualTo(UUIDUtil.convert(UUID_VAL));
+ assertThat(rec.getField("f")).isEqualTo(BYTES_VAL.array());
assertThat(rec.getField("bi")).isEqualTo(BYTES_VAL);
assertThat(rec.getField("li")).isEqualTo(LIST_VAL);
assertThat(rec.getField("ma")).isEqualTo(MAP_VAL);
+
+ // check by actually writing it
+ for (String format : new String[] {"parquet"}) {
+ IcebergSinkConfig config = mock(IcebergSinkConfig.class);
+ when(config.tableConfig(any())).thenReturn(mock(TableSinkConfig.class));
+ when(config.writeProps()).thenReturn(ImmutableMap.of("write.format.default", format));
+ WriteResult result = writeTest(ImmutableList.of(rec), config, UnpartitionedWriter.class);
+
+ assertThat(result.dataFiles()).hasSize(1);
+ assertThat(result.dataFiles())
+ .allMatch(file -> file.format() == FileFormat.fromString(format));
+ assertThat(result.deleteFiles()).hasSize(0);
Review Comment:
This tests fails before the fix for parquet and aparently fixing parquet it made ORC fail due to UUID handling differently
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#issuecomment-2438754792
Thanks @singhpk234 for the PR and @jbonofre, @bryanck and @ajantha-bhat For Review!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1811314119
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java:
##########
@@ -126,10 +127,14 @@ private Object convertValue(
case STRING:
return convertString(value);
case UUID:
+ if ("parquet".equals(config.writeProps().get("write.format.default"))) {
Review Comment:
TableProperties.DEFAULT_FILE_FORMAT
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "ajantha-bhat (via GitHub)" <gi...@apache.org>.
ajantha-bhat commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1811920666
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java:
##########
@@ -206,11 +227,16 @@ public static void beforeAll() {
@BeforeEach
public void before() {
+ super.before();
+ ImmutableMap<String, String> mp = ImmutableMap.of("write.format.default", format);
Review Comment:
Use TableProperties.DEFAULT_FILE_FORMAT
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "bryanck (via GitHub)" <gi...@apache.org>.
bryanck commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1811331398
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java:
##########
@@ -126,10 +127,14 @@ private Object convertValue(
case STRING:
return convertString(value);
case UUID:
+ if ("parquet".equals(config.writeProps().get("write.format.default"))) {
+ return UUIDUtil.convert(convertUUID(value));
+ }
return convertUUID(value);
case BINARY:
- case FIXED:
return convertBase64Binary(value);
+ case FIXED:
+ return convertBase64Binary(value).array();
Review Comment:
+1 for `ByteBuffers.toByteArray()`. The buffer shouldn't be reused, a record is converted to an Iceberg row, written, then dereferenced. I don't believe any of the writers cache rows as-is.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1811341710
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java:
##########
@@ -126,10 +127,14 @@ private Object convertValue(
case STRING:
return convertString(value);
case UUID:
+ if ("parquet".equals(config.writeProps().get("write.format.default"))) {
+ return UUIDUtil.convert(convertUUID(value));
Review Comment:
sounds good
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1815763222
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java:
##########
@@ -84,11 +93,18 @@
import org.apache.kafka.connect.storage.ConverterType;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
-public class RecordConverterTest {
+@ExtendWith(ParameterizedTestExtension.class)
+public class RecordConverterTest extends BaseWriterTest {
Review Comment:
Sure, added this test, Thanks for suggesting this !
I also do think we need an E2E with write I can take this test as a followup to this pr as it would require refactoring of the writer tests.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#issuecomment-2432596526
This change looks good, I'm just wondering about the test. I would have kept the original test and create a new one dedicated for the writer.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "bryanck (via GitHub)" <gi...@apache.org>.
bryanck commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1811313113
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java:
##########
@@ -84,11 +93,18 @@
import org.apache.kafka.connect.storage.ConverterType;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
-public class RecordConverterTest {
+@ExtendWith(ParameterizedTestExtension.class)
+public class RecordConverterTest extends BaseWriterTest {
Review Comment:
I was hoping we'd keep this test specific to the conversion functions, and keep writer tests separate. Do you have thoughts on that?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1811295504
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java:
##########
@@ -126,10 +127,14 @@ private Object convertValue(
case STRING:
return convertString(value);
case UUID:
+ if ("parquet".equals(config.writeProps().get("write.format.default"))) {
+ return UUIDUtil.convert(convertUUID(value));
+ }
return convertUUID(value);
case BINARY:
- case FIXED:
return convertBase64Binary(value);
+ case FIXED:
+ return convertBase64Binary(value).array();
Review Comment:
ByteBuffers.toByteArray(...)
Also just making sure, we won't be re-using this buffer anywhere?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [KafkaConnect] Fix RecordConverter for UUID and Fixed Types [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #11346:
URL: https://github.com/apache/iceberg/pull/11346#discussion_r1806695581
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/RecordConverterTest.java:
##########
@@ -921,11 +929,24 @@ private void assertRecordValues(Record record) {
assertThat(rec.getField("dec")).isEqualTo(DEC_VAL);
assertThat(rec.getField("s")).isEqualTo(STR_VAL);
assertThat(rec.getField("b")).isEqualTo(true);
- assertThat(rec.getField("u")).isEqualTo(UUID_VAL);
- assertThat(rec.getField("f")).isEqualTo(BYTES_VAL);
+ assertThat(rec.getField("u")).isEqualTo(UUIDUtil.convert(UUID_VAL));
+ assertThat(rec.getField("f")).isEqualTo(BYTES_VAL.array());
assertThat(rec.getField("bi")).isEqualTo(BYTES_VAL);
assertThat(rec.getField("li")).isEqualTo(LIST_VAL);
assertThat(rec.getField("ma")).isEqualTo(MAP_VAL);
+
+ // check by actually writing it
+ for (String format : new String[] {"parquet"}) {
+ IcebergSinkConfig config = mock(IcebergSinkConfig.class);
+ when(config.tableConfig(any())).thenReturn(mock(TableSinkConfig.class));
+ when(config.writeProps()).thenReturn(ImmutableMap.of("write.format.default", format));
+ WriteResult result = writeTest(ImmutableList.of(rec), config, UnpartitionedWriter.class);
+
+ assertThat(result.dataFiles()).hasSize(1);
+ assertThat(result.dataFiles())
+ .allMatch(file -> file.format() == FileFormat.fromString(format));
+ assertThat(result.deleteFiles()).hasSize(0);
Review Comment:
This tests fails before the fix for parquet.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org