This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new d80f51417 [arrow] Enable upper-case field name in arrow format writer
(#4263)
d80f51417 is described below
commit d80f51417fad76c43f5cf531575b9d17b33593ee
Author: YeJunHao <[email protected]>
AuthorDate: Thu Sep 26 15:19:27 2024 +0800
[arrow] Enable upper-case field name in arrow format writer (#4263)
---
.../src/main/java/org/apache/paimon/arrow/ArrowBundleRecords.java | 7 +++++--
.../paimon/arrow/converter/Arrow2PaimonVectorConverter.java | 4 ++--
.../java/org/apache/paimon/arrow/reader/ArrowBatchReader.java | 8 ++++++--
.../java/org/apache/paimon/arrow/vector/ArrowFormatWriter.java | 5 +++++
.../org/apache/paimon/arrow/vector/ArrowFormatWriterTest.java | 8 ++++----
.../paimon/arrow/vector/OneElementFieldVectorGeneratorTest.java | 6 +++---
6 files changed, 25 insertions(+), 13 deletions(-)
diff --git
a/paimon-arrow/src/main/java/org/apache/paimon/arrow/ArrowBundleRecords.java
b/paimon-arrow/src/main/java/org/apache/paimon/arrow/ArrowBundleRecords.java
index 073672c75..84dfb875e 100644
--- a/paimon-arrow/src/main/java/org/apache/paimon/arrow/ArrowBundleRecords.java
+++ b/paimon-arrow/src/main/java/org/apache/paimon/arrow/ArrowBundleRecords.java
@@ -32,10 +32,13 @@ public class ArrowBundleRecords implements BundleRecords {
private final VectorSchemaRoot vectorSchemaRoot;
private final RowType rowType;
+ private final boolean onlyLowerCase;
- public ArrowBundleRecords(VectorSchemaRoot vectorSchemaRoot, RowType
rowType) {
+ public ArrowBundleRecords(
+ VectorSchemaRoot vectorSchemaRoot, RowType rowType, boolean
onlyLowerCase) {
this.vectorSchemaRoot = vectorSchemaRoot;
this.rowType = rowType;
+ this.onlyLowerCase = onlyLowerCase;
}
public VectorSchemaRoot getVectorSchemaRoot() {
@@ -49,7 +52,7 @@ public class ArrowBundleRecords implements BundleRecords {
@Override
public Iterator<InternalRow> iterator() {
- ArrowBatchReader arrowBatchReader = new ArrowBatchReader(rowType);
+ ArrowBatchReader arrowBatchReader = new ArrowBatchReader(rowType,
onlyLowerCase);
return arrowBatchReader.readBatch(vectorSchemaRoot).iterator();
}
}
diff --git
a/paimon-arrow/src/main/java/org/apache/paimon/arrow/converter/Arrow2PaimonVectorConverter.java
b/paimon-arrow/src/main/java/org/apache/paimon/arrow/converter/Arrow2PaimonVectorConverter.java
index 6536ae924..2bcaab50b 100644
---
a/paimon-arrow/src/main/java/org/apache/paimon/arrow/converter/Arrow2PaimonVectorConverter.java
+++
b/paimon-arrow/src/main/java/org/apache/paimon/arrow/converter/Arrow2PaimonVectorConverter.java
@@ -390,7 +390,7 @@ public interface Arrow2PaimonVectorConverter {
return Timestamp.fromMicros(value);
} else {
return Timestamp.fromEpochMillis(
- value / 1_000_000, (int) value %
1_000_000);
+ value / 1_000_000, (int) (value %
1_000_000));
}
}
};
@@ -417,7 +417,7 @@ public interface Arrow2PaimonVectorConverter {
return Timestamp.fromMicros(value);
} else {
return Timestamp.fromEpochMillis(
- value / 1_000_000, (int) value %
1_000_000);
+ value / 1_000_000, (int) (value %
1_000_000));
}
}
};
diff --git
a/paimon-arrow/src/main/java/org/apache/paimon/arrow/reader/ArrowBatchReader.java
b/paimon-arrow/src/main/java/org/apache/paimon/arrow/reader/ArrowBatchReader.java
index 364bf4884..a349dfff0 100644
---
a/paimon-arrow/src/main/java/org/apache/paimon/arrow/reader/ArrowBatchReader.java
+++
b/paimon-arrow/src/main/java/org/apache/paimon/arrow/reader/ArrowBatchReader.java
@@ -41,8 +41,9 @@ public class ArrowBatchReader {
private final VectorizedColumnBatch batch;
private final Arrow2PaimonVectorConverter[] convertors;
private final RowType projectedRowType;
+ private final boolean toLowerCase;
- public ArrowBatchReader(RowType rowType) {
+ public ArrowBatchReader(RowType rowType, boolean toLowerCase) {
this.internalRowSerializer = new InternalRowSerializer(rowType);
ColumnVector[] columnVectors = new
ColumnVector[rowType.getFieldCount()];
this.convertors = new
Arrow2PaimonVectorConverter[rowType.getFieldCount()];
@@ -52,6 +53,7 @@ public class ArrowBatchReader {
for (int i = 0; i < columnVectors.length; i++) {
this.convertors[i] =
Arrow2PaimonVectorConverter.construct(rowType.getTypeAt(i));
}
+ this.toLowerCase = toLowerCase;
}
public Iterable<InternalRow> readBatch(VectorSchemaRoot vsr) {
@@ -60,7 +62,9 @@ public class ArrowBatchReader {
List<DataField> dataFields = projectedRowType.getFields();
for (int i = 0; i < dataFields.size(); ++i) {
try {
- Field field =
arrowSchema.findField(dataFields.get(i).name().toLowerCase());
+ String fieldName = dataFields.get(i).name();
+ Field field =
+ arrowSchema.findField(toLowerCase ?
fieldName.toLowerCase() : fieldName);
int idx = arrowSchema.getFields().indexOf(field);
mapping[i] = idx;
} catch (IllegalArgumentException e) {
diff --git
a/paimon-arrow/src/main/java/org/apache/paimon/arrow/vector/ArrowFormatWriter.java
b/paimon-arrow/src/main/java/org/apache/paimon/arrow/vector/ArrowFormatWriter.java
index dde2500d8..acdb5d0dc 100644
---
a/paimon-arrow/src/main/java/org/apache/paimon/arrow/vector/ArrowFormatWriter.java
+++
b/paimon-arrow/src/main/java/org/apache/paimon/arrow/vector/ArrowFormatWriter.java
@@ -27,10 +27,14 @@ import org.apache.paimon.types.RowType;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.util.OversizedAllocationException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/** Write from {@link InternalRow} to {@link VectorSchemaRoot}. */
public class ArrowFormatWriter implements AutoCloseable {
+ private static final Logger LOG =
LoggerFactory.getLogger(ArrowFormatWriter.class);
+
private final VectorSchemaRoot vectorSchemaRoot;
private final ArrowFieldWriter[] fieldWriters;
@@ -72,6 +76,7 @@ public class ArrowFormatWriter implements AutoCloseable {
fieldWriters[i].write(rowId, currentRow, i);
} catch (OversizedAllocationException | IndexOutOfBoundsException
e) {
// maybe out of memory
+ LOG.warn("Arrow field writer failed while writing", e);
return false;
}
}
diff --git
a/paimon-arrow/src/test/java/org/apache/paimon/arrow/vector/ArrowFormatWriterTest.java
b/paimon-arrow/src/test/java/org/apache/paimon/arrow/vector/ArrowFormatWriterTest.java
index 5892616f1..0b6cb7eca 100644
---
a/paimon-arrow/src/test/java/org/apache/paimon/arrow/vector/ArrowFormatWriterTest.java
+++
b/paimon-arrow/src/test/java/org/apache/paimon/arrow/vector/ArrowFormatWriterTest.java
@@ -100,7 +100,7 @@ public class ArrowFormatWriterTest {
writer.flush();
VectorSchemaRoot vectorSchemaRoot = writer.getVectorSchemaRoot();
- ArrowBatchReader arrowBatchReader = new
ArrowBatchReader(PRIMITIVE_TYPE);
+ ArrowBatchReader arrowBatchReader = new
ArrowBatchReader(PRIMITIVE_TYPE, true);
Iterable<InternalRow> rows =
arrowBatchReader.readBatch(vectorSchemaRoot);
Iterator<InternalRow> iterator = rows.iterator();
@@ -142,7 +142,7 @@ public class ArrowFormatWriterTest {
}
vectors.set(vectors.size() - 1, vector0);
- ArrowBatchReader arrowBatchReader = new
ArrowBatchReader(PRIMITIVE_TYPE);
+ ArrowBatchReader arrowBatchReader = new
ArrowBatchReader(PRIMITIVE_TYPE, true);
Iterable<InternalRow> rows = arrowBatchReader.readBatch(new
VectorSchemaRoot(vectors));
Iterator<InternalRow> iterator = rows.iterator();
@@ -177,7 +177,7 @@ public class ArrowFormatWriterTest {
VectorSchemaRoot vectorSchemaRoot = writer.getVectorSchemaRoot();
Iterator<InternalRow> iterator =
- new ArrowBundleRecords(vectorSchemaRoot,
PRIMITIVE_TYPE).iterator();
+ new ArrowBundleRecords(vectorSchemaRoot, PRIMITIVE_TYPE,
true).iterator();
for (int i = 0; i < 1000; i++) {
InternalRow actual = iterator.next();
InternalRow expectec = list.get(i);
@@ -208,7 +208,7 @@ public class ArrowFormatWriterTest {
writer.flush();
VectorSchemaRoot vectorSchemaRoot = writer.getVectorSchemaRoot();
- ArrowBatchReader arrowBatchReader = new
ArrowBatchReader(PRIMITIVE_TYPE);
+ ArrowBatchReader arrowBatchReader = new
ArrowBatchReader(PRIMITIVE_TYPE, true);
Iterable<InternalRow> rows =
arrowBatchReader.readBatch(vectorSchemaRoot);
Iterator<InternalRow> iterator = rows.iterator();
diff --git
a/paimon-arrow/src/test/java/org/apache/paimon/arrow/vector/OneElementFieldVectorGeneratorTest.java
b/paimon-arrow/src/test/java/org/apache/paimon/arrow/vector/OneElementFieldVectorGeneratorTest.java
index e9b672e8f..a417133da 100644
---
a/paimon-arrow/src/test/java/org/apache/paimon/arrow/vector/OneElementFieldVectorGeneratorTest.java
+++
b/paimon-arrow/src/test/java/org/apache/paimon/arrow/vector/OneElementFieldVectorGeneratorTest.java
@@ -49,7 +49,7 @@ public class OneElementFieldVectorGeneratorTest {
new OneElementFieldVectorGenerator(rootAllocator,
dataField, value);
try (FieldVector fieldVector =
oneElementFieldVectorGenerator.get(10000)) {
ArrowBatchReader reader =
- new ArrowBatchReader(new
RowType(Arrays.asList(dataField)));
+ new ArrowBatchReader(new
RowType(Arrays.asList(dataField)), true);
Iterable<InternalRow> it =
reader.readBatch(new
VectorSchemaRoot(Arrays.asList(fieldVector)));
it.forEach(
@@ -68,7 +68,7 @@ public class OneElementFieldVectorGeneratorTest {
new OneElementFieldVectorGenerator(rootAllocator,
dataField, value)) {
FieldVector fieldVector =
oneElementFieldVectorGenerator.get(10000);
ArrowBatchReader reader =
- new ArrowBatchReader(new
RowType(Arrays.asList(dataField)));
+ new ArrowBatchReader(new
RowType(Arrays.asList(dataField)), true);
Iterable<InternalRow> it =
reader.readBatch(new
VectorSchemaRoot(Arrays.asList(fieldVector)));
it.forEach(i ->
Assertions.assertThat(i.getInt(0)).isEqualTo(genericRow.getInt(0)));
@@ -85,7 +85,7 @@ public class OneElementFieldVectorGeneratorTest {
try (FieldVector fieldVector =
oneElementFieldVectorGenerator.get(100000)) {
Assertions.assertThat(fieldVector.getValueCount()).isEqualTo(100000);
ArrowBatchReader reader =
- new ArrowBatchReader(new
RowType(Arrays.asList(dataField)));
+ new ArrowBatchReader(new
RowType(Arrays.asList(dataField)), true);
Iterable<InternalRow> it =
reader.readBatch(new
VectorSchemaRoot(Arrays.asList(fieldVector)));
it.forEach(
