You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "manolama (via GitHub)" <gi...@apache.org> on 2023/10/23 21:17:55 UTC
[PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
manolama opened a new pull request, #38423:
URL: https://github.com/apache/arrow/pull/38423
<!--
Thanks for opening a pull request!
If this is your first pull request you can find detailed information on how
to contribute here:
* [New Contributor's Guide](https://arrow.apache.org/docs/dev/developers/guide/step_by_step/pr_lifecycle.html#reviews-and-merge-of-the-pull-request)
* [Contributing Overview](https://arrow.apache.org/docs/dev/developers/overview.html)
If this is not a [minor PR](https://github.com/apache/arrow/blob/main/CONTRIBUTING.md#Minor-Fixes). Could you open an issue for this pull request on GitHub? https://github.com/apache/arrow/issues/new/choose
Opening GitHub issues ahead of time contributes to the [Openness](http://theapacheway.com/open/#:~:text=Openness%20allows%20new%20users%20the,must%20happen%20in%20the%20open.) of the Apache Arrow project.
Then could you also rename the pull request title in the following format?
GH-${GITHUB_ISSUE_ID}: [${COMPONENT}] ${SUMMARY}
or
MINOR: [${COMPONENT}] ${SUMMARY}
In the case of PARQUET issues on JIRA the title also supports:
PARQUET-${JIRA_ISSUE_ID}: [${COMPONENT}] ${SUMMARY}
-->
### Rationale for this change
Java is missing delta encoding for dictionaries per https://arrow.apache.org/docs/status.html. This PR will implement delta encoding and a helper to easily write dictionary encoded columns in batched streams and files.
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
### What changes are included in this PR?
Adds a new `BaseDictionary` interface that the original `Dictionary` implements. A new `BatchedDictionary` is added that allows for writing data to the dictionary in batches that are then flushed to an IPC stream or file. On flush, the `BatchedDictionary` is reset to support delta encoding or replacement (for streams only).
Also adds a flag to the `DictionaryEncoding` to enable delta encoding.
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
### Are these changes tested?
Yes, multiple unit tests on IPC files and streams. Not tested with add-on libraries.
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
### Are there any user-facing changes?
Yes with new APIs. Existing APIs are backwards compatible so no code changes are needed.
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please uncomment the line below and explain which changes are breaking.
-->
<!-- **This PR includes breaking changes to public APIs.** -->
<!--
Please uncomment the line below (and provide explanation) if the changes fix either (a) a security vulnerability, (b) a bug that caused incorrect or invalid data to be produced, or (c) a bug that causes a crash (even when the API contract is upheld). We use this to highlight fixes to issues that may affect users without their knowledge. For this reason, fixing bugs that cause errors don't count, since those are usually obvious.
-->
<!-- **This PR contains a "Critical Fix".** -->
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1778493012
@manolama All CI's are passing now, I will take a look.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385882139
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
Review Comment:
Let me know re https://github.com/apache/arrow/pull/38423#discussion_r1385845189 and I can tweak this.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "davisusanibar (via GitHub)" <gi...@apache.org>.
davisusanibar commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1419645300
##########
java/vector/src/test/java/org/apache/arrow/vector/dictionary/TestBatchedDictionary.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 org.apache.arrow.vector.dictionary;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.UInt2Vector;
+import org.apache.arrow.vector.ValueVector;
+import org.apache.arrow.vector.types.FloatingPointPrecision;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class TestBatchedDictionary {
+
+ private static final DictionaryEncoding DELTA =
+ new DictionaryEncoding(42, false, new ArrowType.Int(16, false), true);
+ private static final DictionaryEncoding SINGLE =
+ new DictionaryEncoding(24, false, new ArrowType.Int(16, false));
+ private static final byte[] FOO = "foo".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] BAR = "bar".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] BAZ = "baz".getBytes(StandardCharsets.UTF_8);
+
+ private BufferAllocator allocator;
+
+ private static List<ArrowType> validDictionaryTypes = Arrays.asList(
+ new ArrowType.Utf8(),
+ ArrowType.Binary.INSTANCE
+ );
+ private static List<ArrowType> invalidDictionaryTypes = Arrays.asList(
+ new ArrowType.LargeUtf8(),
+ new ArrowType.LargeBinary(),
+ new ArrowType.Bool(),
+ new ArrowType.Int(8, false)
+ );
+ private static List<ArrowType> validIndexTypes = Arrays.asList(
+ new ArrowType.Int(16, false),
+ new ArrowType.Int(8, false),
+ new ArrowType.Int(32, false),
+ new ArrowType.Int(64, false),
+ new ArrowType.Int(16, true),
+ new ArrowType.Int(8, true),
+ new ArrowType.Int(32, true),
+ new ArrowType.Int(64, true)
+ );
+ private static List<ArrowType> invalidIndexTypes = Arrays.asList(
+ new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE),
+ new ArrowType.Bool(),
+ new ArrowType.Utf8()
+ );
+
+ @BeforeEach
+ public void prepare() {
+ allocator = new RootAllocator(1024 * 1024);
+ }
+
+ @AfterEach
+ public void shutdown() {
+ allocator.close();
+ }
+
+ public static Collection<Arguments> validTypes() {
+ List<Arguments> params = new ArrayList<>();
+ for (ArrowType dictType : validDictionaryTypes) {
+ for (ArrowType indexType : validIndexTypes) {
+ params.add(Arguments.arguments(dictType, indexType));
+ }
+ }
+ return params;
+ }
+
+ public static Collection<Arguments> invalidTypes() {
+ List<Arguments> params = new ArrayList<>();
+ for (ArrowType dictType : invalidDictionaryTypes) {
+ for (ArrowType indexType : validIndexTypes) {
+ params.add(Arguments.arguments(dictType, indexType));
+ }
+ }
+ for (ArrowType dictType : validDictionaryTypes) {
+ for (ArrowType indexType : invalidIndexTypes) {
+ params.add(Arguments.arguments(dictType, indexType));
+ }
+ }
+ return params;
+ }
+
+ @ParameterizedTest
+ @MethodSource("validTypes")
+ public void testValidDictionaryTypes(ArrowType dictType, ArrowType indexType) throws IOException {
+ new BatchedDictionary(
+ "vector",
+ DELTA,
+ dictType,
+ indexType,
+ allocator
+ ).close();
+ }
Review Comment:
To apply validations, does assert not need to be added?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1384294051
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
+
+ private int deltaIndex;
+
+ private int dictionaryIndex;
+
+ private boolean wasReset;
Review Comment:
What does this variable do? Just for the clarity in naming.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385269406
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
Review Comment:
Hmm, how about `isFile`? The goal is to have a hint from the user so we can throw an exception if they accidentally try to write a replacement to a stream. And since the file IPC format seems to be rarely used, I'd rather have it "default" to streaming mode.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385883271
##########
java/vector/src/test/java/org/apache/arrow/vector/ipc/TestArrowFile.java:
##########
@@ -131,4 +150,67 @@ public void testFileStreamHasEos() throws IOException {
}
}
}
+
+ @ParameterizedTest
+ @MethodSource("dictionaryParams")
+ public void testMultiBatchDictionaries(int state) throws Exception {
+ File file = new File("target/mytest_multi_batch_dictionaries_" + state + ".arrow");
Review Comment:
good point, I was just copying previous tests. I'll file an issue.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1791875605
@manolama sorry I couldn't review it yet. I will allocate some time today and Monday for this.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1459548098
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/DictionaryProvider.java:
##########
@@ -79,15 +84,24 @@ public final Set<Long> getDictionaryIds() {
}
@Override
- public Dictionary lookup(long id) {
+ public BaseDictionary lookup(long id) {
return map.get(id);
}
@Override
public void close() {
- for (Dictionary dictionary : map.values()) {
+ for (BaseDictionary dictionary : map.values()) {
Review Comment:
Ack, thanks.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385848843
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/DictionaryProvider.java:
##########
@@ -79,15 +84,24 @@ public final Set<Long> getDictionaryIds() {
}
@Override
- public Dictionary lookup(long id) {
+ public BaseDictionary lookup(long id) {
return map.get(id);
}
@Override
public void close() {
- for (Dictionary dictionary : map.values()) {
+ for (BaseDictionary dictionary : map.values()) {
dictionary.getVector().close();
}
}
+
+ @Override
+ public void resetBatchedDictionaries() {
Review Comment:
Should I just make a generic reset API with the `Dictionary` having a no-op reset?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1843744091
| @vibhatha are you able to re-review this?
@lidavidm There is one more thing I have to look at (tied up lately). The integration tests are generating dictionaries in streams with multiple batches then writing those streams to files in each language. Java now throws when it detects the replacement dictionary. I need to go through the other code bases and see if they are simply dropping the replacements silently at write or if they are indeed writing the replacements and violating spec.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1843733146
> > One thing that would be really great to have alongside this change is an update to the docs https://arrow.apache.org/docs/java/vector.html#dictionary-encoding. Creating an issue to update the docs after merging works, too.
>
> Aye, I wanted to add that and some cookbook entries as well. Are the docs updated only on a release so that if I submit a PR and it's merged, it won't go out until v14?
There are docs that reflect the development version as well.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1424823464
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileWriter.java:
##########
@@ -130,14 +130,24 @@ protected void endInternal(WriteChannel out) throws IOException {
protected void ensureDictionariesWritten(DictionaryProvider provider, Set<Long> dictionaryIdsUsed)
throws IOException {
if (dictionariesWritten) {
+ for (long id : dictionaryIdsUsed) {
+ BaseDictionary dictionary = provider.lookup(id);
+ if (dictionary.getEncoding().isDelta()) {
+ writeDictionaryBatch(dictionary, false);
+ } else if (dictionary.mark() > 0) {
+ throw new IllegalStateException("Replacement dictionaries are not supported in the " +
+ "IPC file format. Dictionary ID: " + dictionary.getEncoding().getId());
+ }
+ }
return;
}
+
dictionariesWritten = true;
// Write out all dictionaries required.
// Replacement dictionaries are not supported in the IPC file format.
for (long id : dictionaryIdsUsed) {
- Dictionary dictionary = provider.lookup(id);
- writeDictionaryBatch(dictionary);
+ BaseDictionary dictionary = provider.lookup(id);
+ writeDictionaryBatch(dictionary, true);
Review Comment:
nit: may be comment the variable name for `true`
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1459566905
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileWriter.java:
##########
@@ -130,14 +130,24 @@ protected void endInternal(WriteChannel out) throws IOException {
protected void ensureDictionariesWritten(DictionaryProvider provider, Set<Long> dictionaryIdsUsed)
throws IOException {
if (dictionariesWritten) {
+ for (long id : dictionaryIdsUsed) {
+ BaseDictionary dictionary = provider.lookup(id);
+ if (dictionary.getEncoding().isDelta()) {
+ writeDictionaryBatch(dictionary, false);
+ } else if (dictionary.mark() > 0) {
+ throw new IllegalStateException("Replacement dictionaries are not supported in the " +
+ "IPC file format. Dictionary ID: " + dictionary.getEncoding().getId());
+ }
+ }
return;
}
+
dictionariesWritten = true;
// Write out all dictionaries required.
// Replacement dictionaries are not supported in the IPC file format.
for (long id : dictionaryIdsUsed) {
- Dictionary dictionary = provider.lookup(id);
- writeDictionaryBatch(dictionary);
+ BaseDictionary dictionary = provider.lookup(id);
+ writeDictionaryBatch(dictionary, true);
Review 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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1459550999
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileReader.java:
##########
@@ -140,21 +137,6 @@ public Map<String, String> getMetaData() {
return new HashMap<>();
}
- /**
- * Read a dictionary batch from the source, will be invoked after the schema has been read and
- * called N times, where N is the number of dictionaries indicated by the schema Fields.
- *
- * @return the read ArrowDictionaryBatch
- * @throws IOException on error
- */
- public ArrowDictionaryBatch readDictionary() throws IOException {
Review Comment:
Ah, good catch, thanks.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support [arrow]
Posted by "davisusanibar (via GitHub)" <gi...@apache.org>.
davisusanibar commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1460194781
##########
java/vector/src/test/java/org/apache/arrow/vector/ipc/TestArrowStream.java:
##########
@@ -64,7 +82,7 @@ public void testStreamZeroLengthBatch() throws IOException {
try (IntVector vector = new IntVector("foo", allocator);) {
Schema schema = new Schema(Collections.singletonList(vector.getField()));
try (VectorSchemaRoot root =
- new VectorSchemaRoot(schema, Collections.singletonList(vector), vector.getValueCount());
+ new VectorSchemaRoot(schema, Collections.singletonList(vector), vector.getValueCount());
Review Comment:
> CC @danepitkin @davisusanibar I think it would help contributors if we used a formatter like google-java-format instead of peppering them with questions like this that checkstyle apparently doesn't handle
I am currently working on PR #39712 to enable the `google-java-format` by using the spotless plugin.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1457125005
##########
java/vector/src/test/java/org/apache/arrow/vector/ipc/TestArrowStream.java:
##########
@@ -64,7 +82,7 @@ public void testStreamZeroLengthBatch() throws IOException {
try (IntVector vector = new IntVector("foo", allocator);) {
Schema schema = new Schema(Collections.singletonList(vector.getField()));
try (VectorSchemaRoot root =
- new VectorSchemaRoot(schema, Collections.singletonList(vector), vector.getValueCount());
+ new VectorSchemaRoot(schema, Collections.singletonList(vector), vector.getValueCount());
Review Comment:
Did you mean to make this change?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1459565891
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileReader.java:
##########
@@ -164,12 +146,59 @@ public boolean loadNextBatch() throws IOException {
ArrowBlock block = footer.getRecordBatches().get(currentRecordBatch++);
ArrowRecordBatch batch = readRecordBatch(in, block, allocator);
loadRecordBatch(batch);
+ try {
+ loadDictionaries();
+ } catch (IOException e) {
+ batch.close();
+ throw e;
+ }
return true;
} else {
return false;
}
}
+ /**
+ * Loads any dictionaries that may be needed by the given record batch. It attempts
+ * to read as little as possible but may read in more deltas than are necessary for blocks
+ * toward the end of the file.
+ */
+ private void loadDictionaries() throws IOException {
+ // initial load
+ if (currentDictionaryBatch == 0) {
+ for (int i = 0; i < dictionaries.size(); i++) {
+ ArrowBlock block = footer.getDictionaries().get(currentDictionaryBatch++);
+ ArrowDictionaryBatch dictionaryBatch = readDictionaryBatch(in, block, allocator);
+ loadDictionary(dictionaryBatch, false);
+ }
+ estimatedDictionaryRecordBatch++;
+ } else {
+ // we need to look for delta dictionaries. It involves a look-ahead, unfortunately.
+ HashSet<Long> visited = new HashSet<Long>();
+ while (estimatedDictionaryRecordBatch < currentRecordBatch &&
+ currentDictionaryBatch < footer.getDictionaries().size()) {
+ ArrowBlock block = footer.getDictionaries().get(currentDictionaryBatch++);
+ ArrowDictionaryBatch dictionaryBatch = readDictionaryBatch(in, block, allocator);
+ long dictionaryId = dictionaryBatch.getDictionaryId();
+ if (visited.contains(dictionaryId)) {
+ // done
+ currentDictionaryBatch--;
+ estimatedDictionaryRecordBatch++;
+ } else if (!dictionaries.containsKey(dictionaryId)) {
+ throw new IOException("Dictionary ID " + dictionaryId + " was written " +
+ "after the initial batch. The file does not follow the IPC file protocol.");
+ } else if (!dictionaryBatch.isDelta()) {
+ throw new IOException("Dictionary ID " + dictionaryId + " was written as a replacement " +
+ "after the initial batch. Replacement dictionaries are not currently allowed in the IPC file protocol.");
+ } else {
+ loadDictionary(dictionaryBatch, true);
+ }
+ }
+ if (currentDictionaryBatch >= footer.getDictionaries().size()) {
Review Comment:
Yup, removed, leftover cruft. Thanks!
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "jduo (via GitHub)" <gi...@apache.org>.
jduo commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1394478488
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/DictionaryProvider.java:
##########
@@ -79,15 +84,24 @@ public final Set<Long> getDictionaryIds() {
}
@Override
- public Dictionary lookup(long id) {
+ public BaseDictionary lookup(long id) {
return map.get(id);
}
@Override
public void close() {
- for (Dictionary dictionary : map.values()) {
+ for (BaseDictionary dictionary : map.values()) {
Review Comment:
It makes sense to let the creator close. It should be documented though. The Java codebase is somewhat inconsistent about ownership unfortunately.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385848500
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/DictionaryProvider.java:
##########
@@ -79,15 +84,24 @@ public final Set<Long> getDictionaryIds() {
}
@Override
- public Dictionary lookup(long id) {
+ public BaseDictionary lookup(long id) {
return map.get(id);
}
@Override
public void close() {
- for (Dictionary dictionary : map.values()) {
+ for (BaseDictionary dictionary : map.values()) {
Review Comment:
Good question, what model of ownership does the Java impl follow? In the batch dictionary I offered a means of passing in two vectors in which case I'd assume the creator would be responsible for closing. Maybe I should remove that and if folks need nested dictionaries later we offer a ctor with a parent vector? Then the dictionary could be closed 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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385998927
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
Review Comment:
My argument is mainly on the naming. I am fine with the current change.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1384395189
##########
java/memory/memory-core/src/test/java/org/apache/arrow/memory/util/TestArrowBufPointer.java:
##########
@@ -204,6 +204,11 @@ public int hashCode(ArrowBuf buf, long offset, long length) {
return SimpleHasher.INSTANCE.hashCode(buf, offset, length);
}
+ @Override
+ public int hashCode(byte[] buf, int offset, int length) {
+ return 0;
Review Comment:
could you please explain a bit why this returns 0?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1384438762
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileWriter.java:
##########
@@ -130,14 +130,24 @@ protected void endInternal(WriteChannel out) throws IOException {
protected void ensureDictionariesWritten(DictionaryProvider provider, Set<Long> dictionaryIdsUsed)
throws IOException {
if (dictionariesWritten) {
+ for (long id : dictionaryIdsUsed) {
+ BaseDictionary dictionary = provider.lookup(id);
+ if (dictionary.getEncoding().isDelta()) {
+ writeDictionaryBatch(dictionary, false);
+ }
+ // TODO - It would be useful to throw an exception here if a replacement dictionary was found
+ // with modifications. Replacements are not currently allowed in files. For now, we just drop it
+ // and throw an exception in the BatchedDictionary and hope users use that class.
+ }
Review Comment:
Shall we create an issue for this?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385884655
##########
java/vector/src/test/java/org/apache/arrow/vector/ipc/BaseFileTest.java:
##########
@@ -846,4 +862,293 @@ protected void validateListAsMapData(VectorSchemaRoot root) {
}
}
}
+
+ /**
+ * Utility to write permutations of dictionary encoding.
+ *
+ * state == 1, one delta dictionary.
Review Comment:
yeah, easy enough.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1776246022
@manolama there are 3 CI's failing and the error is consistent.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1384386091
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowReader.java:
##########
@@ -243,6 +248,8 @@ protected void loadDictionary(ArrowDictionaryBatch dictionaryBatch) {
}
return;
}
+ // TODO - If the super class is a file reader it may be good to throw an exception here
+ // the dictionary failed to satisfy the spec (i.e. being a replacement dictionary)
Review Comment:
It would be better if we could create an issue for this.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1457817130
##########
java/memory/memory-core/src/test/java/org/apache/arrow/memory/util/TestArrowBufPointer.java:
##########
@@ -204,6 +204,11 @@ public int hashCode(ArrowBuf buf, long offset, long length) {
return SimpleHasher.INSTANCE.hashCode(buf, offset, length);
}
+ @Override
+ public int hashCode(byte[] buf, int offset, int length) {
+ throw new UnsupportedOperationException("Not used in UT.");
Review Comment:
Unit Test.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1459599450
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/Dictionary.java:
##########
@@ -72,4 +74,14 @@ public boolean equals(Object o) {
public int hashCode() {
return Objects.hash(encoding, dictionary);
}
+
+ @Override
+ public int mark() {
+ return reset == true ? 0 : dictionary.getValueCount();
+ }
+
+ @Override
+ public void reset() {
+ reset = true;
+ }
Review Comment:
I'm more comfortable with adding it to the provider. You could even introduce it as a subclass or wrapper/proxy class that the writer checks for? Then the base provider doesn't support replacement/deltas and the new provider would let you explicitly update dictionaries.
I also don't like Arrow-Java's tendency towards mutable state/data. It makes things like this messy since we can't rely on object identity, unfortunately.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "jduo (via GitHub)" <gi...@apache.org>.
jduo commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1394479902
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
+
+ private int deltaIndex;
+
+ private int dictionaryIndex;
+
+ private boolean wasReset;
+
+ /**
+ * Creates a dictionary with two vectors of the given types. The dictionary vector
+ * will be named "{name}-dictionary".
+ * <p>
+ * To use this dictionary, provide the dictionary vector to a {@link DictionaryProvider},
+ * add the {@link #getIndexVector()} to the {@link org.apache.arrow.vector.VectorSchemaRoot}
+ * and call the {@link #setSafe(int, byte[], int, int)} or other set methods.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator
+ ) {
+ this(name, encoding, dictionaryType, indexType, forFileIPC, allocator, "-dictionary");
+ }
+
+ /**
+ * Creates a dictionary with two vectors of the given types.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ * @param suffix A non-null suffix to append to the name of the dictionary.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator,
+ String suffix
+ ) {
+ this.encoding = encoding;
+ this.forFileIPC = forFileIPC;
+ FieldVector vector = new FieldType(false, dictionaryType, null)
+ .createNewSingleVector(name + suffix, allocator, null);
+ if (!(BaseVariableWidthVector.class.isAssignableFrom(vector.getClass()))) {
+ throw new IllegalArgumentException("Dictionary must be a superclass of 'BaseVariableWidthVector' " +
+ "such as 'VarCharVector'.");
+ }
+ dictionary = (BaseVariableWidthVector) vector;
+ vector = new FieldType(true, indexType, encoding)
+ .createNewSingleVector(name, allocator, null);
+ if (!(BaseIntVector.class.isAssignableFrom(vector.getClass()))) {
+ throw new IllegalArgumentException("Index vector must be a superclass type of 'BaseIntVector' " +
+ "such as 'IntVector' or 'Uint4Vector'.");
+ }
+ indexVector = (BaseIntVector) vector;
+ hashTable = new DictionaryHashTable();
+ }
+
+ /**
+ * Creates a dictionary that will populate the provided vectors with data. Useful if
+ * dictionaries need to be children of a parent vector.
+ * @param dictionary The dictionary to hold the original data.
+ * @param indexVector The index to store the encoded offsets.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ */
+ public BatchedDictionary(
+ FieldVector dictionary,
+ FieldVector indexVector,
+ boolean forFileIPC
+ ) {
+ this.encoding = dictionary.getField().getDictionary();
+ this.forFileIPC = forFileIPC;
+ if (!(BaseVariableWidthVector.class.isAssignableFrom(dictionary.getClass()))) {
+ throw new IllegalArgumentException("Dictionary must be a superclass of 'BaseVariableWidthVector' " +
+ "such as 'VarCharVector'.");
+ }
+ if (dictionary.getField().isNullable()) {
+ throw new IllegalArgumentException("Dictionary must be non-nullable.");
+ }
+ this.dictionary = (BaseVariableWidthVector) dictionary;
+ if (!(BaseIntVector.class.isAssignableFrom(indexVector.getClass()))) {
+ throw new IllegalArgumentException("Index vector must be a superclass type of 'BaseIntVector' " +
+ "such as 'IntVector' or 'Uint4Vector'.");
+ }
+ this.indexVector = (BaseIntVector) indexVector;
+ hashTable = new DictionaryHashTable();
+ }
+
+ /**
+ * The index vector.
+ */
+ public FieldVector getIndexVector() {
+ return indexVector;
+ }
+
+ @Override
+ public FieldVector getVector() {
+ return dictionary;
+ }
+
+ @Override
+ public ArrowType getVectorType() {
+ return dictionary.getField().getType();
+ }
+
+ @Override
+ public DictionaryEncoding getEncoding() {
+ return encoding;
+ }
+
+ /**
+ * Considers the entire byte array as the dictionary value. If the value is null,
+ * a null will be written to the index.
+ *
+ * @param index the value to change
+ * @param value the value to write.
+ */
+ public void setSafe(int index, byte[] value) {
+ if (value == null) {
+ setNull(index);
+ return;
+ }
+ setSafe(index, value, 0, value.length);
+ }
+
+ /**
+ * Encodes the given range in the dictionary. If the value is null, a null will be
+ * written to the index.
+ *
+ * @param index the value to change
+ * @param value the value to write.
+ * @param offset An offset into the value array.
+ * @param len The length of the value to write.
+ */
+ public void setSafe(int index, byte[] value, int offset, int len) {
+ if (value == null || len == 0) {
+ setNull(index);
+ return;
+ }
+ int di = getIndex(value, offset, len);
+ indexVector.setWithPossibleTruncate(index, di);
+ }
+
+ /**
+ * Set the element at the given index to null.
+ *
+ * @param index the value to change
+ */
+ public void setNull(int index) {
+ indexVector.setNull(index);
+ }
+
+ @Override
+ public void close() throws IOException {
+ dictionary.close();
+ indexVector.close();
+ }
+
+ /**
+ * Mark the dictionary as complete for the batch. Called by the {@link ArrowWriter}
+ * on {@link ArrowWriter#writeBatch()}.
+ */
+ public void mark() {
+ dictionary.setValueCount(dictionaryIndex);
+ // not setting the index vector value count. That will happen when the user calls
+ // VectorSchemaRoot#setRowCount().
+ }
+
+ /**
+ * Resets the dictionary to be used for a new batch. Called by the {@link ArrowWriter} on
+ * {@link ArrowWriter#writeBatch()}.
+ */
+ public void reset() {
+ wasReset = true;
Review Comment:
I guess when I see a "reset" function I think it's to restore an object to an initial state so that it can be re-used.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1424814658
##########
java/memory/memory-core/src/test/java/org/apache/arrow/memory/util/TestArrowBufPointer.java:
##########
@@ -204,6 +204,11 @@ public int hashCode(ArrowBuf buf, long offset, long length) {
return SimpleHasher.INSTANCE.hashCode(buf, offset, length);
}
+ @Override
+ public int hashCode(byte[] buf, int offset, int length) {
+ throw new UnsupportedOperationException("Not used in UT.");
Review Comment:
What UT stands for?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1843732584
@vibhatha are you able to re-review this?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support [arrow]
Posted by "kou (via GitHub)" <gi...@apache.org>.
kou commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-2502368172
We're moving `java/` to https://github.com/apache/arrow-java .
Could you open a new PR in apache/arrow-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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "danepitkin (via GitHub)" <gi...@apache.org>.
danepitkin commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385614601
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
+
+ private int deltaIndex;
+
+ private int dictionaryIndex;
+
+ private boolean wasReset;
Review Comment:
If its not required, I would consider deleting it to reduce state managed inside this object.
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowReader.java:
##########
@@ -243,6 +248,8 @@ protected void loadDictionary(ArrowDictionaryBatch dictionaryBatch) {
}
return;
}
+ // TODO - If the super class is a file reader it may be good to throw an exception here
+ // the dictionary failed to satisfy the spec (i.e. being a replacement dictionary)
Review Comment:
Can we enforce this in the reader (instead of a todo?)
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
Review Comment:
Could we instead have the IPC stream interface check and fail if a replacement dictionary is sent? I think it would be nice if we don't maintain this state inside the dictionary itself.
##########
java/memory/memory-core/src/main/java/org/apache/arrow/memory/util/hash/MurmurHasher.java:
##########
@@ -106,6 +111,36 @@ public static int hashCode(long address, long length, int seed) {
return finalizeHashCode(hash, length);
}
+ /**
+ * Calculates the hash code for a byte array.
+ * @param buffer the non-null buffer to read.
+ * @param offset an offset into the byte array.
+ * @param length length of the memory region.
+ * @param seed the seed.
+ * @return the hash code.
+ */
+ public static int hashCode(byte[] buffer, int offset, int length, int seed) {
Review Comment:
Why are we adding support for `byte[]` buffers instead of using `ArrowBuf`?
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileWriter.java:
##########
@@ -130,14 +130,24 @@ protected void endInternal(WriteChannel out) throws IOException {
protected void ensureDictionariesWritten(DictionaryProvider provider, Set<Long> dictionaryIdsUsed)
throws IOException {
if (dictionariesWritten) {
+ for (long id : dictionaryIdsUsed) {
+ BaseDictionary dictionary = provider.lookup(id);
+ if (dictionary.getEncoding().isDelta()) {
+ writeDictionaryBatch(dictionary, false);
+ }
+ // TODO - It would be useful to throw an exception here if a replacement dictionary was found
+ // with modifications. Replacements are not currently allowed in files. For now, we just drop it
+ // and throw an exception in the BatchedDictionary and hope users use that class.
+ }
Review Comment:
I'd prefer we enforce this here in the writer than in the dictionary 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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "davisusanibar (via GitHub)" <gi...@apache.org>.
davisusanibar commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1419063634
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowReader.java:
##########
@@ -242,6 +247,9 @@ protected void loadDictionary(ArrowDictionaryBatch dictionaryBatch) {
VectorBatchAppender.batchAppend(vector, deltaVector);
}
return;
+ } else if (validateReplacements && getClass() == ArrowFileReader.class) {
+ throw new IllegalStateException("Replacement dictionaries are not supported in " +
+ "the IPC file format. Dictionary ID: " + dictionary.getEncoding().getId());
Review Comment:
Would it be possible to include these new changes/updates in the IPC documentation file -> docs/source/java/ipc.rst#reading-writing-ipc-formats?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1384289084
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
Review Comment:
nit: would it be better to name this as `isStream`?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1459081736
##########
java/vector/src/main/java/org/apache/arrow/vector/types/pojo/DictionaryEncoding.java:
##########
@@ -65,6 +81,11 @@ public Int getIndexType() {
return indexType;
}
+ @JsonGetter("isDelta")
+ public boolean isDelta() {
Review Comment:
So this means: if true, the associated Dictionary is a delta dictionary for a previous Dictionary with the same ID.
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileReader.java:
##########
@@ -140,21 +137,6 @@ public Map<String, String> getMetaData() {
return new HashMap<>();
}
- /**
- * Read a dictionary batch from the source, will be invoked after the schema has been read and
- * called N times, where N is the number of dictionaries indicated by the schema Fields.
- *
- * @return the read ArrowDictionaryBatch
- * @throws IOException on error
- */
- public ArrowDictionaryBatch readDictionary() throws IOException {
Review Comment:
We're going to need to mark this as a breaking change.
##########
java/vector/src/main/java/org/apache/arrow/vector/types/pojo/DictionaryEncoding.java:
##########
@@ -42,14 +43,29 @@ public class DictionaryEncoding {
* @param indexType (nullable). The integer type to use for indexing in the dictionary. Defaults to a signed
* 32 bit integer.
*/
+ public DictionaryEncoding(long id, boolean ordered, Int indexType) {
+ this(id, ordered, indexType, false);
+ }
+
+ /**
+ * Constructs a new instance.
+ *
+ * @param id The ID of the dictionary to use for encoding.
+ * @param ordered Whether the keys in values in the dictionary are ordered.
+ * @param indexType (nullable). The integer type to use for indexing in the dictionary. Defaults to a signed
+ * 32 bit integer.
+ * @param isDelta Whether the dictionary is a delta dictionary.
+ */
@JsonCreator
public DictionaryEncoding(
@JsonProperty("id") long id,
@JsonProperty("isOrdered") boolean ordered,
- @JsonProperty("indexType") Int indexType) {
+ @JsonProperty("indexType") Int indexType,
+ @JsonProperty("isDelta") Boolean isDelta) {
Review Comment:
Does this really need boxing? It should always be true or false.
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BaseDictionary.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+
+/**
+ * Base interface for various dictionary implementations. Implementations include
+ * {@link Dictionary} for encoding a complete vector and {@link BatchedDictionary}
+ * for continuous encoding of a vector.
+ * These methods provide means of accessing the dictionary vector containing the
+ * encoded data.
+ */
+public interface BaseDictionary {
+
+ /**
+ * The dictionary vector containing unique entries.
+ */
+ FieldVector getVector();
+
+ /**
+ * The encoding used for the dictionary vector.
+ */
+ DictionaryEncoding getEncoding();
+
+ /**
+ * The type of the dictionary vector.
+ */
+ ArrowType getVectorType();
+
+ /**
+ * Mark the dictionary as complete for the batch. Called by the {@link ArrowWriter}
+ * on {@link ArrowWriter#writeBatch()}.
+ * @return The number of values written to the dictionary.
+ */
+ int mark();
Review Comment:
nit: we may want to prefer long for future-proofness? Though perhaps it'll be too painful with the current APIs
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowReader.java:
##########
@@ -242,6 +247,9 @@ protected void loadDictionary(ArrowDictionaryBatch dictionaryBatch) {
VectorBatchAppender.batchAppend(vector, deltaVector);
}
return;
+ } else if (validateReplacements && getClass() == ArrowFileReader.class) {
+ throw new IllegalStateException("Replacement dictionaries are not supported in " +
+ "the IPC file format. Dictionary ID: " + dictionary.getEncoding().getId());
Review Comment:
This is better as a separate issue. But yes, others have noted that the language about what is and is not allowed is very confusing.
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation for continuous encoding of data in a dictionary and
+ * index vector as opposed to the {@link Dictionary} that encodes a complete vector.
+ * Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private int deltaIndex;
+
+ private int dictionaryIndex;
+
+ private boolean oneTimeEncoding;
+
+ private boolean wasReset;
+
+ private boolean vectorsProvided;
+
+ /**
+ * Creates a dictionary and index vector with the respective types. The dictionary vector
+ * will be named "{name}-dictionary".
+ * <p>
+ * To use this dictionary, provide the dictionary vector to a {@link DictionaryProvider},
+ * add the {@link #getIndexVector()} to the {@link org.apache.arrow.vector.VectorSchemaRoot}
+ * and call the {@link #setSafe(int, byte[], int, int)} or other set methods.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param allocator The allocator to use.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ BufferAllocator allocator
+ ) {
+ this(name, encoding, dictionaryType, indexType, allocator, "-dictionary");
+ }
+
+ /**
+ * Creates a dictionary index vector with the respective types.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param allocator The allocator to use.
+ * @param suffix A non-null suffix to append to the name of the dictionary.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ BufferAllocator allocator,
+ String suffix
+ ) {
+ this(name, encoding, dictionaryType, indexType, allocator, suffix, false);
+ }
+
+ /**
+ * Creates a dictionary index vector with the respective types.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param allocator The allocator to use.
+ * @param suffix A non-null suffix to append to the name of the dictionary.
+ * @param oneTimeEncoding A mode where the entries can be added to the dictionary until
+ * the first stream batch is written. After that, any new entries
+ * to the dictionary will throw an exception.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ BufferAllocator allocator,
+ String suffix,
+ boolean oneTimeEncoding
+ ) {
+ this.encoding = encoding;
+ this.oneTimeEncoding = oneTimeEncoding;
+ if (dictionaryType.getTypeID() != ArrowType.ArrowTypeID.Utf8 &&
+ dictionaryType.getTypeID() != ArrowType.ArrowTypeID.Binary) {
Review Comment:
Dictionary value types are not limited by the spec. I think it's OK to limit support for an initial implementation and file issues (linked to from the source) for expansion.
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/Dictionary.java:
##########
@@ -72,4 +74,14 @@ public boolean equals(Object o) {
public int hashCode() {
return Objects.hash(encoding, dictionary);
}
+
+ @Override
+ public int mark() {
+ return reset == true ? 0 : dictionary.getValueCount();
+ }
+
+ @Override
+ public void reset() {
+ reset = true;
+ }
Review Comment:
I guess what makes me uneasy overall is that we have introduced mutable state in the Dictionary itself to track things needed for writing, and I think instead it would be better to try and keep that state contained in the writer itself as it was before. Also, I agree that 'mark' and 'reset' are somewhat confusing names, though I suppose they make sense in the context of 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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1459748118
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/Dictionary.java:
##########
@@ -72,4 +74,14 @@ public boolean equals(Object o) {
public int hashCode() {
return Objects.hash(encoding, dictionary);
}
+
+ @Override
+ public int mark() {
+ return reset == true ? 0 : dictionary.getValueCount();
+ }
+
+ @Override
+ public void reset() {
+ reset = true;
+ }
Review Comment:
Ack, I'll re-work it as a subclass, thanks! Yup, immutable objects would be nice for this case.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "jduo (via GitHub)" <gi...@apache.org>.
jduo commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1394476407
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/DictionaryProvider.java:
##########
@@ -79,15 +84,24 @@ public final Set<Long> getDictionaryIds() {
}
@Override
- public Dictionary lookup(long id) {
+ public BaseDictionary lookup(long id) {
return map.get(id);
}
@Override
public void close() {
- for (Dictionary dictionary : map.values()) {
+ for (BaseDictionary dictionary : map.values()) {
dictionary.getVector().close();
}
}
+
+ @Override
+ public void resetBatchedDictionaries() {
Review Comment:
I think so. It would be more consistent.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1781764049
I'm also taking a look. Thanks !
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1384395189
##########
java/memory/memory-core/src/test/java/org/apache/arrow/memory/util/TestArrowBufPointer.java:
##########
@@ -204,6 +204,11 @@ public int hashCode(ArrowBuf buf, long offset, long length) {
return SimpleHasher.INSTANCE.hashCode(buf, offset, length);
}
+ @Override
+ public int hashCode(byte[] buf, int offset, int length) {
+ return 0;
Review Comment:
could you please explain a bit why this returns `0`?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1424822503
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileReader.java:
##########
@@ -164,12 +146,59 @@ public boolean loadNextBatch() throws IOException {
ArrowBlock block = footer.getRecordBatches().get(currentRecordBatch++);
ArrowRecordBatch batch = readRecordBatch(in, block, allocator);
loadRecordBatch(batch);
+ try {
+ loadDictionaries();
+ } catch (IOException e) {
+ batch.close();
+ throw e;
+ }
return true;
} else {
return false;
}
}
+ /**
+ * Loads any dictionaries that may be needed by the given record batch. It attempts
+ * to read as little as possible but may read in more deltas than are necessary for blocks
+ * toward the end of the file.
+ */
+ private void loadDictionaries() throws IOException {
+ // initial load
+ if (currentDictionaryBatch == 0) {
+ for (int i = 0; i < dictionaries.size(); i++) {
+ ArrowBlock block = footer.getDictionaries().get(currentDictionaryBatch++);
+ ArrowDictionaryBatch dictionaryBatch = readDictionaryBatch(in, block, allocator);
+ loadDictionary(dictionaryBatch, false);
+ }
+ estimatedDictionaryRecordBatch++;
+ } else {
+ // we need to look for delta dictionaries. It involves a look-ahead, unfortunately.
+ HashSet<Long> visited = new HashSet<Long>();
+ while (estimatedDictionaryRecordBatch < currentRecordBatch &&
+ currentDictionaryBatch < footer.getDictionaries().size()) {
+ ArrowBlock block = footer.getDictionaries().get(currentDictionaryBatch++);
+ ArrowDictionaryBatch dictionaryBatch = readDictionaryBatch(in, block, allocator);
+ long dictionaryId = dictionaryBatch.getDictionaryId();
+ if (visited.contains(dictionaryId)) {
+ // done
+ currentDictionaryBatch--;
+ estimatedDictionaryRecordBatch++;
+ } else if (!dictionaries.containsKey(dictionaryId)) {
+ throw new IOException("Dictionary ID " + dictionaryId + " was written " +
+ "after the initial batch. The file does not follow the IPC file protocol.");
+ } else if (!dictionaryBatch.isDelta()) {
+ throw new IOException("Dictionary ID " + dictionaryId + " was written as a replacement " +
+ "after the initial batch. Replacement dictionaries are not currently allowed in the IPC file protocol.");
+ } else {
+ loadDictionary(dictionaryBatch, true);
+ }
+ }
+ if (currentDictionaryBatch >= footer.getDictionaries().size()) {
Review Comment:
Also,
```java
while (estimatedDictionaryRecordBatch < currentRecordBatch &&
currentDictionaryBatch < footer.getDictionaries().size())
```
doesn't the while handle this case?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385882993
##########
java/vector/src/test/java/org/apache/arrow/vector/ipc/BaseFileTest.java:
##########
@@ -846,4 +862,293 @@ protected void validateListAsMapData(VectorSchemaRoot root) {
}
}
}
+
+ /**
+ * Utility to write permutations of dictionary encoding.
+ *
+ * state == 1, one delta dictionary.
+ * state == 2, one standalone dictionary.
+ * state == 3, one of each
+ * state == 4, delta with nothing at start and end
+ * state == 5, both deltas
+ * state == 6, both deltas and standalone
+ * state == 7, replacement dictionary
+ */
+ protected void writeDataMultiBatchWithDictionaries(OutputStream stream, int state) throws IOException {
+ DictionaryProvider.MapDictionaryProvider provider = new DictionaryProvider.MapDictionaryProvider();
+ DictionaryEncoding deltaEncoding =
+ new DictionaryEncoding(42, false, new ArrowType.Int(16, false), true);
+ DictionaryEncoding replacementEncoding =
+ new DictionaryEncoding(24, false, new ArrowType.Int(16, false), false);
+ DictionaryEncoding deltaCEncoding =
+ new DictionaryEncoding(1, false, new ArrowType.Int(16, false), true);
+ DictionaryEncoding replacementEncodingUpdated =
+ new DictionaryEncoding(2, false, new ArrowType.Int(16, false), false);
+
+ boolean isFile = stream instanceof FileOutputStream;
+ try (BatchedDictionary vectorA = newDictionary("vectorA", deltaEncoding, isFile);
+ BatchedDictionary vectorB = newDictionary("vectorB", replacementEncoding, isFile);
+ BatchedDictionary vectorC = newDictionary("vectorC", deltaCEncoding, isFile);
+ BatchedDictionary vectorD = newDictionary("vectorD", replacementEncodingUpdated, isFile);) {
+ switch (state) {
+ case 1:
+ provider.put(vectorA);
+ break;
+ case 2:
+ provider.put(vectorB);
+ break;
+ case 3:
+ provider.put(vectorA);
+ provider.put(vectorB);
+ break;
+ case 4:
+ provider.put(vectorC);
+ break;
+ case 5:
+ provider.put(vectorA);
+ provider.put(vectorC);
+ break;
+ case 6:
+ provider.put(vectorA);
+ provider.put(vectorB);
+ provider.put(vectorC);
+ break;
+ case 7:
+ provider.put(vectorD);
+ break;
+ default:
+ throw new IllegalStateException("Unsupported state: " + state);
+ }
+
+ VectorSchemaRoot root = null;
+ switch (state) {
+ case 1:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector());
+ break;
+ case 2:
+ root = VectorSchemaRoot.of(vectorB.getIndexVector());
+ break;
+ case 3:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector(), vectorB.getIndexVector());
+ break;
+ case 4:
+ root = VectorSchemaRoot.of(vectorC.getIndexVector());
+ break;
+ case 5:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector(), vectorC.getIndexVector());
+ break;
+ case 6:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector(), vectorB.getIndexVector(), vectorC.getIndexVector());
+ break;
+ case 7:
+ root = VectorSchemaRoot.of(vectorD.getIndexVector());
+ break;
+ default:
+ throw new IllegalStateException("Unsupported state: " + state);
+ }
+
+ ArrowWriter arrowWriter = null;
+ try {
+ if (stream instanceof FileOutputStream) {
+ FileChannel channel = ((FileOutputStream) stream).getChannel();
+ arrowWriter = new ArrowFileWriter(root, provider, channel);
+ } else {
+ arrowWriter = new ArrowStreamWriter(root, provider, stream);
+ }
+
+ vectorA.setSafe(0, "foo".getBytes(StandardCharsets.UTF_8));
+ vectorA.setSafe(1, "bar".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(0, "lorem".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(1, "ipsum".getBytes(StandardCharsets.UTF_8));
+ vectorC.setNull(0);
+ vectorC.setNull(1);
+ vectorD.setSafe(0, "porro".getBytes(StandardCharsets.UTF_8));
+ vectorD.setSafe(1, "amet".getBytes(StandardCharsets.UTF_8));
+
+ // batch 1
+ arrowWriter.start();
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ // batch 2
+ vectorA.setSafe(0, "meep".getBytes(StandardCharsets.UTF_8));
+ vectorA.setSafe(1, "bar".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(0, "ipsum".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(1, "lorem".getBytes(StandardCharsets.UTF_8));
+ vectorC.setSafe(0, "qui".getBytes(StandardCharsets.UTF_8));
+ vectorC.setSafe(1, "dolor".getBytes(StandardCharsets.UTF_8));
+ vectorD.setSafe(0, "amet".getBytes(StandardCharsets.UTF_8));
+ if (state == 7) {
+ vectorD.setSafe(1, "quia".getBytes(StandardCharsets.UTF_8));
+ }
+
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ // batch 3
+ vectorA.setNull(0);
+ vectorA.setNull(1);
+ vectorB.setSafe(0, "ipsum".getBytes(StandardCharsets.UTF_8));
+ vectorB.setNull(1);
+ vectorC.setNull(0);
+ vectorC.setSafe(1, "qui".getBytes(StandardCharsets.UTF_8));
+ vectorD.setNull(0);
+ if (state == 7) {
+ vectorD.setSafe(1, "quia".getBytes(StandardCharsets.UTF_8));
+ }
+
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ // batch 4
+ vectorA.setSafe(0, "bar".getBytes(StandardCharsets.UTF_8));
+ vectorA.setSafe(1, "zap".getBytes(StandardCharsets.UTF_8));
+ vectorB.setNull(0);
+ vectorB.setSafe(1, "lorem".getBytes(StandardCharsets.UTF_8));
+ vectorC.setNull(0);
+ vectorC.setNull(1);
+ if (state == 7) {
+ vectorD.setSafe(0, "quia".getBytes(StandardCharsets.UTF_8));
+ }
+ vectorD.setNull(1);
+
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ arrowWriter.end();
+ } catch (Exception e) {
+ if (arrowWriter != null) {
+ arrowWriter.close();
+ }
+ throw e;
+ }
+ }
+ }
+
+ Map<Integer, String[][]> valuesPerBlock = new HashMap<Integer, String[][]>();
+
+ {
+ valuesPerBlock.put(0, new String[][]{
+ {"foo", "bar"},
+ {"lorem", "ipsum"},
+ {null, null},
+ {"porro", "amet"}
+ });
+ valuesPerBlock.put(1, new String[][]{
+ {"meep", "bar"},
+ {"ipsum", "lorem"},
+ {"qui", "dolor"},
+ {"amet", "quia"}
+ });
+ valuesPerBlock.put(2, new String[][]{
+ {null, null},
+ {"ipsum", null},
+ {null, "qui"},
+ {null, "quia"}
+ });
+ valuesPerBlock.put(3, new String[][]{
+ {"bar", "zap"},
+ {null, "lorem"},
+ {null, null},
+ {"quia", null}
+ });
+ }
+
+ protected void assertDictionary(FieldVector encoded, ArrowReader reader, String... expected) throws Exception {
+ DictionaryEncoding dictionaryEncoding = encoded.getField().getDictionary();
+ BaseDictionary dictionary = reader.getDictionaryVectors().get(dictionaryEncoding.getId());
+ try (ValueVector decoded = DictionaryEncoder.decode(encoded, dictionary)) {
+ Assertions.assertEquals(expected.length, encoded.getValueCount());
+ for (int i = 0; i < expected.length; i++) {
+ if (expected[i] == null) {
+ Assertions.assertNull(decoded.getObject(i));
+ } else {
+ assertNotNull(decoded.getObject(i));
+ Assertions.assertEquals(expected[i], decoded.getObject(i).toString());
+ }
+ }
+ }
+ }
+
+ protected void assertBlock(File file, int block, int state) throws Exception {
+ try (FileInputStream fileInputStream = new FileInputStream(file);
+ ArrowFileReader reader = new ArrowFileReader(fileInputStream.getChannel(), allocator);) {
+ reader.loadRecordBatch(reader.getRecordBlocks().get(block));
+ assertBlock(reader, block, state);
+ }
+ }
+
+ protected void assertBlock(ArrowReader reader, int block, int state) throws Exception {
+ VectorSchemaRoot r = reader.getVectorSchemaRoot();
+ FieldVector dictA = r.getVector("vectorA");
+ FieldVector dictB = r.getVector("vectorB");
+ FieldVector dictC = r.getVector("vectorC");
+ FieldVector dictD = r.getVector("vectorD");
+
+ switch (state) {
+ case 1:
+ assertDictionary(dictA, reader, valuesPerBlock.get(block)[0]);
+ assertNull(dictB);
+ assertNull(dictC);
+ assertNull(dictD);
+ break;
+ case 2:
+ assertNull(dictA);
+ assertDictionary(dictB, reader, valuesPerBlock.get(block)[1]);
+ assertNull(dictC);
+ assertNull(dictD);
+ break;
+ case 3:
+ assertDictionary(dictA, reader, valuesPerBlock.get(block)[0]);
+ assertDictionary(dictB, reader, valuesPerBlock.get(block)[1]);
+ assertNull(dictC);
+ assertNull(dictD);
+ break;
+ case 4:
+ assertNull(dictA);
+ assertNull(dictB);
+ assertDictionary(dictC, reader, valuesPerBlock.get(block)[2]);
+ assertNull(dictD);
+ break;
+ case 5:
+ assertDictionary(dictA, reader, valuesPerBlock.get(block)[0]);
+ assertNull(dictB);
+ assertDictionary(dictC, reader, valuesPerBlock.get(block)[2]);
+ assertNull(dictD);
+ break;
+ case 6:
+ assertDictionary(dictA, reader, valuesPerBlock.get(block)[0]);
+ assertDictionary(dictB, reader, valuesPerBlock.get(block)[1]);
+ assertDictionary(dictC, reader, valuesPerBlock.get(block)[2]);
+ assertNull(dictD);
+ break;
+ case 7:
+ assertNull(dictA);
+ assertNull(dictB);
+ assertNull(dictC);
+ assertDictionary(dictD, reader, valuesPerBlock.get(block)[3]);
+ break;
+ default:
+ throw new IllegalStateException("Unsupported state: " + state);
+ }
+ }
+
+ protected static Collection<Arguments> dictionaryParams() {
+ List<Arguments> params = new ArrayList<>();
+ for (int i = 1; i < 8; i++) {
Review Comment:
Just the number of unique test states. I'll drop a note.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1800909178
> One thing that would be really great to have alongside this change is an update to the docs https://arrow.apache.org/docs/java/vector.html#dictionary-encoding. Creating an issue to update the docs after merging works, too.
Aye, I wanted to add that and some cookbook entries as well. Are the docs updated only on a release so that if I submit a PR and it's merged, it won't go out until v14?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385840032
##########
java/memory/memory-core/src/main/java/org/apache/arrow/memory/util/hash/MurmurHasher.java:
##########
@@ -106,6 +111,36 @@ public static int hashCode(long address, long length, int seed) {
return finalizeHashCode(hash, length);
}
+ /**
+ * Calculates the hash code for a byte array.
+ * @param buffer the non-null buffer to read.
+ * @param offset an offset into the byte array.
+ * @param length length of the memory region.
+ * @param seed the seed.
+ * @return the hash code.
+ */
+ public static int hashCode(byte[] buffer, int offset, int length, int seed) {
Review Comment:
This is for data that hasn't been written into an `ArrowBuf` via the `Vector.set()` APIs yet. In the dictionary map we can hash the original data and only copy it into an `ArrowBuf` if the hash wasn't present in the map.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1459564658
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileReader.java:
##########
@@ -164,12 +146,59 @@ public boolean loadNextBatch() throws IOException {
ArrowBlock block = footer.getRecordBatches().get(currentRecordBatch++);
ArrowRecordBatch batch = readRecordBatch(in, block, allocator);
loadRecordBatch(batch);
+ try {
+ loadDictionaries();
+ } catch (IOException e) {
+ batch.close();
+ throw e;
+ }
return true;
} else {
return false;
}
}
+ /**
+ * Loads any dictionaries that may be needed by the given record batch. It attempts
+ * to read as little as possible but may read in more deltas than are necessary for blocks
+ * toward the end of the file.
+ */
+ private void loadDictionaries() throws IOException {
+ // initial load
+ if (currentDictionaryBatch == 0) {
+ for (int i = 0; i < dictionaries.size(); i++) {
+ ArrowBlock block = footer.getDictionaries().get(currentDictionaryBatch++);
+ ArrowDictionaryBatch dictionaryBatch = readDictionaryBatch(in, block, allocator);
+ loadDictionary(dictionaryBatch, false);
+ }
+ estimatedDictionaryRecordBatch++;
+ } else {
+ // we need to look for delta dictionaries. It involves a look-ahead, unfortunately.
+ HashSet<Long> visited = new HashSet<Long>();
+ while (estimatedDictionaryRecordBatch < currentRecordBatch &&
+ currentDictionaryBatch < footer.getDictionaries().size()) {
+ ArrowBlock block = footer.getDictionaries().get(currentDictionaryBatch++);
+ ArrowDictionaryBatch dictionaryBatch = readDictionaryBatch(in, block, allocator);
+ long dictionaryId = dictionaryBatch.getDictionaryId();
+ if (visited.contains(dictionaryId)) {
Review Comment:
Doh, left over cruft. thanks.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1776262805
@manolama is [this](https://github.com/apache/arrow/pull/38169#issuecomment-1765339201) related?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1778478547
@vibhatha Finally got it! Please take a look, let me know if the API and changes look ok. Thanks.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385840745
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
+
+ private int deltaIndex;
+
+ private int dictionaryIndex;
+
+ private boolean wasReset;
+
+ /**
+ * Creates a dictionary with two vectors of the given types. The dictionary vector
+ * will be named "{name}-dictionary".
+ * <p>
+ * To use this dictionary, provide the dictionary vector to a {@link DictionaryProvider},
+ * add the {@link #getIndexVector()} to the {@link org.apache.arrow.vector.VectorSchemaRoot}
+ * and call the {@link #setSafe(int, byte[], int, int)} or other set methods.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator
+ ) {
+ this(name, encoding, dictionaryType, indexType, forFileIPC, allocator, "-dictionary");
+ }
+
+ /**
+ * Creates a dictionary with two vectors of the given types.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ * @param suffix A non-null suffix to append to the name of the dictionary.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator,
+ String suffix
+ ) {
+ this.encoding = encoding;
+ this.forFileIPC = forFileIPC;
+ FieldVector vector = new FieldType(false, dictionaryType, null)
+ .createNewSingleVector(name + suffix, allocator, null);
+ if (!(BaseVariableWidthVector.class.isAssignableFrom(vector.getClass()))) {
Review Comment:
Yeah, that works.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "jduo (via GitHub)" <gi...@apache.org>.
jduo commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385786095
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
+
+ private int deltaIndex;
+
+ private int dictionaryIndex;
+
+ private boolean wasReset;
+
+ /**
+ * Creates a dictionary with two vectors of the given types. The dictionary vector
+ * will be named "{name}-dictionary".
+ * <p>
+ * To use this dictionary, provide the dictionary vector to a {@link DictionaryProvider},
+ * add the {@link #getIndexVector()} to the {@link org.apache.arrow.vector.VectorSchemaRoot}
+ * and call the {@link #setSafe(int, byte[], int, int)} or other set methods.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator
+ ) {
+ this(name, encoding, dictionaryType, indexType, forFileIPC, allocator, "-dictionary");
+ }
+
+ /**
+ * Creates a dictionary with two vectors of the given types.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ * @param suffix A non-null suffix to append to the name of the dictionary.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator,
+ String suffix
+ ) {
+ this.encoding = encoding;
+ this.forFileIPC = forFileIPC;
+ FieldVector vector = new FieldType(false, dictionaryType, null)
+ .createNewSingleVector(name + suffix, allocator, null);
+ if (!(BaseVariableWidthVector.class.isAssignableFrom(vector.getClass()))) {
Review Comment:
It seems like we should be able to evaluate dictionaryType before creating the vector rather than validating the type constructed afterwards.
##########
java/vector/src/test/java/org/apache/arrow/vector/ipc/TestArrowFile.java:
##########
@@ -131,4 +150,67 @@ public void testFileStreamHasEos() throws IOException {
}
}
}
+
+ @ParameterizedTest
+ @MethodSource("dictionaryParams")
+ public void testMultiBatchDictionaries(int state) throws Exception {
+ File file = new File("target/mytest_multi_batch_dictionaries_" + state + ".arrow");
Review Comment:
This wouldn't work right if the test got built as a JAR. These inputs should probably be moved to test resoucres and opened using getResource(). This could be a separate issue.
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
+
+ private int deltaIndex;
+
+ private int dictionaryIndex;
+
+ private boolean wasReset;
+
+ /**
+ * Creates a dictionary with two vectors of the given types. The dictionary vector
+ * will be named "{name}-dictionary".
+ * <p>
+ * To use this dictionary, provide the dictionary vector to a {@link DictionaryProvider},
+ * add the {@link #getIndexVector()} to the {@link org.apache.arrow.vector.VectorSchemaRoot}
+ * and call the {@link #setSafe(int, byte[], int, int)} or other set methods.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator
+ ) {
+ this(name, encoding, dictionaryType, indexType, forFileIPC, allocator, "-dictionary");
+ }
+
+ /**
+ * Creates a dictionary with two vectors of the given types.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ * @param suffix A non-null suffix to append to the name of the dictionary.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator,
+ String suffix
+ ) {
+ this.encoding = encoding;
+ this.forFileIPC = forFileIPC;
+ FieldVector vector = new FieldType(false, dictionaryType, null)
+ .createNewSingleVector(name + suffix, allocator, null);
+ if (!(BaseVariableWidthVector.class.isAssignableFrom(vector.getClass()))) {
+ throw new IllegalArgumentException("Dictionary must be a superclass of 'BaseVariableWidthVector' " +
+ "such as 'VarCharVector'.");
+ }
+ dictionary = (BaseVariableWidthVector) vector;
+ vector = new FieldType(true, indexType, encoding)
+ .createNewSingleVector(name, allocator, null);
+ if (!(BaseIntVector.class.isAssignableFrom(vector.getClass()))) {
+ throw new IllegalArgumentException("Index vector must be a superclass type of 'BaseIntVector' " +
Review Comment:
It would be good to validate both the dictionaryType and indexType before creating either vectors so that the constructor never gets into a state where it throws with vectors already constructed (they should actually be released with close()).
##########
java/vector/src/test/java/org/apache/arrow/vector/ipc/BaseFileTest.java:
##########
@@ -846,4 +862,293 @@ protected void validateListAsMapData(VectorSchemaRoot root) {
}
}
}
+
+ /**
+ * Utility to write permutations of dictionary encoding.
+ *
+ * state == 1, one delta dictionary.
Review Comment:
Could this be an enum?
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileReader.java:
##########
@@ -164,12 +146,54 @@ public boolean loadNextBatch() throws IOException {
ArrowBlock block = footer.getRecordBatches().get(currentRecordBatch++);
ArrowRecordBatch batch = readRecordBatch(in, block, allocator);
loadRecordBatch(batch);
+ loadDictionaries();
Review Comment:
If loadDictionaries() throws an exception, does the batch leak?
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/DictionaryProvider.java:
##########
@@ -79,15 +84,24 @@ public final Set<Long> getDictionaryIds() {
}
@Override
- public Dictionary lookup(long id) {
+ public BaseDictionary lookup(long id) {
return map.get(id);
}
@Override
public void close() {
- for (Dictionary dictionary : map.values()) {
+ for (BaseDictionary dictionary : map.values()) {
dictionary.getVector().close();
}
}
+
+ @Override
+ public void resetBatchedDictionaries() {
Review Comment:
It is strange that only batched dictionaries need to be reset.
##########
java/vector/src/test/java/org/apache/arrow/vector/ipc/BaseFileTest.java:
##########
@@ -846,4 +862,293 @@ protected void validateListAsMapData(VectorSchemaRoot root) {
}
}
}
+
+ /**
+ * Utility to write permutations of dictionary encoding.
+ *
+ * state == 1, one delta dictionary.
+ * state == 2, one standalone dictionary.
+ * state == 3, one of each
+ * state == 4, delta with nothing at start and end
+ * state == 5, both deltas
+ * state == 6, both deltas and standalone
+ * state == 7, replacement dictionary
+ */
+ protected void writeDataMultiBatchWithDictionaries(OutputStream stream, int state) throws IOException {
+ DictionaryProvider.MapDictionaryProvider provider = new DictionaryProvider.MapDictionaryProvider();
+ DictionaryEncoding deltaEncoding =
+ new DictionaryEncoding(42, false, new ArrowType.Int(16, false), true);
+ DictionaryEncoding replacementEncoding =
+ new DictionaryEncoding(24, false, new ArrowType.Int(16, false), false);
+ DictionaryEncoding deltaCEncoding =
+ new DictionaryEncoding(1, false, new ArrowType.Int(16, false), true);
+ DictionaryEncoding replacementEncodingUpdated =
+ new DictionaryEncoding(2, false, new ArrowType.Int(16, false), false);
+
+ boolean isFile = stream instanceof FileOutputStream;
+ try (BatchedDictionary vectorA = newDictionary("vectorA", deltaEncoding, isFile);
+ BatchedDictionary vectorB = newDictionary("vectorB", replacementEncoding, isFile);
+ BatchedDictionary vectorC = newDictionary("vectorC", deltaCEncoding, isFile);
+ BatchedDictionary vectorD = newDictionary("vectorD", replacementEncodingUpdated, isFile);) {
+ switch (state) {
+ case 1:
+ provider.put(vectorA);
+ break;
+ case 2:
+ provider.put(vectorB);
+ break;
+ case 3:
+ provider.put(vectorA);
+ provider.put(vectorB);
+ break;
+ case 4:
+ provider.put(vectorC);
+ break;
+ case 5:
+ provider.put(vectorA);
+ provider.put(vectorC);
+ break;
+ case 6:
+ provider.put(vectorA);
+ provider.put(vectorB);
+ provider.put(vectorC);
+ break;
+ case 7:
+ provider.put(vectorD);
+ break;
+ default:
+ throw new IllegalStateException("Unsupported state: " + state);
+ }
+
+ VectorSchemaRoot root = null;
+ switch (state) {
+ case 1:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector());
+ break;
+ case 2:
+ root = VectorSchemaRoot.of(vectorB.getIndexVector());
+ break;
+ case 3:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector(), vectorB.getIndexVector());
+ break;
+ case 4:
+ root = VectorSchemaRoot.of(vectorC.getIndexVector());
+ break;
+ case 5:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector(), vectorC.getIndexVector());
+ break;
+ case 6:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector(), vectorB.getIndexVector(), vectorC.getIndexVector());
+ break;
+ case 7:
+ root = VectorSchemaRoot.of(vectorD.getIndexVector());
+ break;
+ default:
+ throw new IllegalStateException("Unsupported state: " + state);
+ }
+
+ ArrowWriter arrowWriter = null;
+ try {
+ if (stream instanceof FileOutputStream) {
+ FileChannel channel = ((FileOutputStream) stream).getChannel();
+ arrowWriter = new ArrowFileWriter(root, provider, channel);
+ } else {
+ arrowWriter = new ArrowStreamWriter(root, provider, stream);
+ }
+
+ vectorA.setSafe(0, "foo".getBytes(StandardCharsets.UTF_8));
+ vectorA.setSafe(1, "bar".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(0, "lorem".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(1, "ipsum".getBytes(StandardCharsets.UTF_8));
+ vectorC.setNull(0);
+ vectorC.setNull(1);
+ vectorD.setSafe(0, "porro".getBytes(StandardCharsets.UTF_8));
+ vectorD.setSafe(1, "amet".getBytes(StandardCharsets.UTF_8));
+
+ // batch 1
+ arrowWriter.start();
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ // batch 2
+ vectorA.setSafe(0, "meep".getBytes(StandardCharsets.UTF_8));
+ vectorA.setSafe(1, "bar".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(0, "ipsum".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(1, "lorem".getBytes(StandardCharsets.UTF_8));
+ vectorC.setSafe(0, "qui".getBytes(StandardCharsets.UTF_8));
+ vectorC.setSafe(1, "dolor".getBytes(StandardCharsets.UTF_8));
+ vectorD.setSafe(0, "amet".getBytes(StandardCharsets.UTF_8));
+ if (state == 7) {
+ vectorD.setSafe(1, "quia".getBytes(StandardCharsets.UTF_8));
+ }
+
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ // batch 3
+ vectorA.setNull(0);
+ vectorA.setNull(1);
+ vectorB.setSafe(0, "ipsum".getBytes(StandardCharsets.UTF_8));
+ vectorB.setNull(1);
+ vectorC.setNull(0);
+ vectorC.setSafe(1, "qui".getBytes(StandardCharsets.UTF_8));
+ vectorD.setNull(0);
+ if (state == 7) {
+ vectorD.setSafe(1, "quia".getBytes(StandardCharsets.UTF_8));
+ }
+
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ // batch 4
+ vectorA.setSafe(0, "bar".getBytes(StandardCharsets.UTF_8));
+ vectorA.setSafe(1, "zap".getBytes(StandardCharsets.UTF_8));
+ vectorB.setNull(0);
+ vectorB.setSafe(1, "lorem".getBytes(StandardCharsets.UTF_8));
+ vectorC.setNull(0);
+ vectorC.setNull(1);
+ if (state == 7) {
+ vectorD.setSafe(0, "quia".getBytes(StandardCharsets.UTF_8));
+ }
+ vectorD.setNull(1);
+
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ arrowWriter.end();
+ } catch (Exception e) {
+ if (arrowWriter != null) {
+ arrowWriter.close();
+ }
+ throw e;
+ }
+ }
+ }
+
+ Map<Integer, String[][]> valuesPerBlock = new HashMap<Integer, String[][]>();
+
+ {
+ valuesPerBlock.put(0, new String[][]{
+ {"foo", "bar"},
+ {"lorem", "ipsum"},
+ {null, null},
+ {"porro", "amet"}
+ });
+ valuesPerBlock.put(1, new String[][]{
+ {"meep", "bar"},
+ {"ipsum", "lorem"},
+ {"qui", "dolor"},
+ {"amet", "quia"}
+ });
+ valuesPerBlock.put(2, new String[][]{
+ {null, null},
+ {"ipsum", null},
+ {null, "qui"},
+ {null, "quia"}
+ });
+ valuesPerBlock.put(3, new String[][]{
+ {"bar", "zap"},
+ {null, "lorem"},
+ {null, null},
+ {"quia", null}
+ });
+ }
+
+ protected void assertDictionary(FieldVector encoded, ArrowReader reader, String... expected) throws Exception {
+ DictionaryEncoding dictionaryEncoding = encoded.getField().getDictionary();
+ BaseDictionary dictionary = reader.getDictionaryVectors().get(dictionaryEncoding.getId());
+ try (ValueVector decoded = DictionaryEncoder.decode(encoded, dictionary)) {
+ Assertions.assertEquals(expected.length, encoded.getValueCount());
+ for (int i = 0; i < expected.length; i++) {
+ if (expected[i] == null) {
+ Assertions.assertNull(decoded.getObject(i));
Review Comment:
It would be good to be consistent about static imports of assertion functions. assertNotNull() is statically imported but assertNull() isn't.
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/DictionaryProvider.java:
##########
@@ -79,15 +84,24 @@ public final Set<Long> getDictionaryIds() {
}
@Override
- public Dictionary lookup(long id) {
+ public BaseDictionary lookup(long id) {
return map.get(id);
}
@Override
public void close() {
- for (Dictionary dictionary : map.values()) {
+ for (BaseDictionary dictionary : map.values()) {
Review Comment:
Does the dictionary itself need to close()?
##########
java/vector/src/test/java/org/apache/arrow/vector/ipc/BaseFileTest.java:
##########
@@ -846,4 +862,293 @@ protected void validateListAsMapData(VectorSchemaRoot root) {
}
}
}
+
+ /**
+ * Utility to write permutations of dictionary encoding.
+ *
+ * state == 1, one delta dictionary.
+ * state == 2, one standalone dictionary.
+ * state == 3, one of each
+ * state == 4, delta with nothing at start and end
+ * state == 5, both deltas
+ * state == 6, both deltas and standalone
+ * state == 7, replacement dictionary
+ */
+ protected void writeDataMultiBatchWithDictionaries(OutputStream stream, int state) throws IOException {
+ DictionaryProvider.MapDictionaryProvider provider = new DictionaryProvider.MapDictionaryProvider();
+ DictionaryEncoding deltaEncoding =
+ new DictionaryEncoding(42, false, new ArrowType.Int(16, false), true);
+ DictionaryEncoding replacementEncoding =
+ new DictionaryEncoding(24, false, new ArrowType.Int(16, false), false);
+ DictionaryEncoding deltaCEncoding =
+ new DictionaryEncoding(1, false, new ArrowType.Int(16, false), true);
+ DictionaryEncoding replacementEncodingUpdated =
+ new DictionaryEncoding(2, false, new ArrowType.Int(16, false), false);
+
+ boolean isFile = stream instanceof FileOutputStream;
+ try (BatchedDictionary vectorA = newDictionary("vectorA", deltaEncoding, isFile);
+ BatchedDictionary vectorB = newDictionary("vectorB", replacementEncoding, isFile);
+ BatchedDictionary vectorC = newDictionary("vectorC", deltaCEncoding, isFile);
+ BatchedDictionary vectorD = newDictionary("vectorD", replacementEncodingUpdated, isFile);) {
+ switch (state) {
+ case 1:
+ provider.put(vectorA);
+ break;
+ case 2:
+ provider.put(vectorB);
+ break;
+ case 3:
+ provider.put(vectorA);
+ provider.put(vectorB);
+ break;
+ case 4:
+ provider.put(vectorC);
+ break;
+ case 5:
+ provider.put(vectorA);
+ provider.put(vectorC);
+ break;
+ case 6:
+ provider.put(vectorA);
+ provider.put(vectorB);
+ provider.put(vectorC);
+ break;
+ case 7:
+ provider.put(vectorD);
+ break;
+ default:
+ throw new IllegalStateException("Unsupported state: " + state);
+ }
+
+ VectorSchemaRoot root = null;
+ switch (state) {
+ case 1:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector());
+ break;
+ case 2:
+ root = VectorSchemaRoot.of(vectorB.getIndexVector());
+ break;
+ case 3:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector(), vectorB.getIndexVector());
+ break;
+ case 4:
+ root = VectorSchemaRoot.of(vectorC.getIndexVector());
+ break;
+ case 5:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector(), vectorC.getIndexVector());
+ break;
+ case 6:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector(), vectorB.getIndexVector(), vectorC.getIndexVector());
+ break;
+ case 7:
+ root = VectorSchemaRoot.of(vectorD.getIndexVector());
+ break;
+ default:
+ throw new IllegalStateException("Unsupported state: " + state);
+ }
+
+ ArrowWriter arrowWriter = null;
+ try {
+ if (stream instanceof FileOutputStream) {
+ FileChannel channel = ((FileOutputStream) stream).getChannel();
+ arrowWriter = new ArrowFileWriter(root, provider, channel);
+ } else {
+ arrowWriter = new ArrowStreamWriter(root, provider, stream);
+ }
+
+ vectorA.setSafe(0, "foo".getBytes(StandardCharsets.UTF_8));
+ vectorA.setSafe(1, "bar".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(0, "lorem".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(1, "ipsum".getBytes(StandardCharsets.UTF_8));
+ vectorC.setNull(0);
+ vectorC.setNull(1);
+ vectorD.setSafe(0, "porro".getBytes(StandardCharsets.UTF_8));
+ vectorD.setSafe(1, "amet".getBytes(StandardCharsets.UTF_8));
+
+ // batch 1
+ arrowWriter.start();
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ // batch 2
+ vectorA.setSafe(0, "meep".getBytes(StandardCharsets.UTF_8));
+ vectorA.setSafe(1, "bar".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(0, "ipsum".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(1, "lorem".getBytes(StandardCharsets.UTF_8));
+ vectorC.setSafe(0, "qui".getBytes(StandardCharsets.UTF_8));
+ vectorC.setSafe(1, "dolor".getBytes(StandardCharsets.UTF_8));
+ vectorD.setSafe(0, "amet".getBytes(StandardCharsets.UTF_8));
+ if (state == 7) {
+ vectorD.setSafe(1, "quia".getBytes(StandardCharsets.UTF_8));
+ }
+
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ // batch 3
+ vectorA.setNull(0);
+ vectorA.setNull(1);
+ vectorB.setSafe(0, "ipsum".getBytes(StandardCharsets.UTF_8));
+ vectorB.setNull(1);
+ vectorC.setNull(0);
+ vectorC.setSafe(1, "qui".getBytes(StandardCharsets.UTF_8));
+ vectorD.setNull(0);
+ if (state == 7) {
+ vectorD.setSafe(1, "quia".getBytes(StandardCharsets.UTF_8));
+ }
+
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ // batch 4
+ vectorA.setSafe(0, "bar".getBytes(StandardCharsets.UTF_8));
+ vectorA.setSafe(1, "zap".getBytes(StandardCharsets.UTF_8));
+ vectorB.setNull(0);
+ vectorB.setSafe(1, "lorem".getBytes(StandardCharsets.UTF_8));
+ vectorC.setNull(0);
+ vectorC.setNull(1);
+ if (state == 7) {
+ vectorD.setSafe(0, "quia".getBytes(StandardCharsets.UTF_8));
+ }
+ vectorD.setNull(1);
+
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ arrowWriter.end();
+ } catch (Exception e) {
+ if (arrowWriter != null) {
+ arrowWriter.close();
+ }
+ throw e;
+ }
+ }
+ }
+
+ Map<Integer, String[][]> valuesPerBlock = new HashMap<Integer, String[][]>();
+
+ {
+ valuesPerBlock.put(0, new String[][]{
+ {"foo", "bar"},
+ {"lorem", "ipsum"},
+ {null, null},
+ {"porro", "amet"}
+ });
+ valuesPerBlock.put(1, new String[][]{
+ {"meep", "bar"},
+ {"ipsum", "lorem"},
+ {"qui", "dolor"},
+ {"amet", "quia"}
+ });
+ valuesPerBlock.put(2, new String[][]{
+ {null, null},
+ {"ipsum", null},
+ {null, "qui"},
+ {null, "quia"}
+ });
+ valuesPerBlock.put(3, new String[][]{
+ {"bar", "zap"},
+ {null, "lorem"},
+ {null, null},
+ {"quia", null}
+ });
+ }
+
+ protected void assertDictionary(FieldVector encoded, ArrowReader reader, String... expected) throws Exception {
+ DictionaryEncoding dictionaryEncoding = encoded.getField().getDictionary();
+ BaseDictionary dictionary = reader.getDictionaryVectors().get(dictionaryEncoding.getId());
+ try (ValueVector decoded = DictionaryEncoder.decode(encoded, dictionary)) {
+ Assertions.assertEquals(expected.length, encoded.getValueCount());
+ for (int i = 0; i < expected.length; i++) {
+ if (expected[i] == null) {
+ Assertions.assertNull(decoded.getObject(i));
+ } else {
+ assertNotNull(decoded.getObject(i));
+ Assertions.assertEquals(expected[i], decoded.getObject(i).toString());
+ }
+ }
+ }
+ }
+
+ protected void assertBlock(File file, int block, int state) throws Exception {
+ try (FileInputStream fileInputStream = new FileInputStream(file);
+ ArrowFileReader reader = new ArrowFileReader(fileInputStream.getChannel(), allocator);) {
+ reader.loadRecordBatch(reader.getRecordBlocks().get(block));
+ assertBlock(reader, block, state);
+ }
+ }
+
+ protected void assertBlock(ArrowReader reader, int block, int state) throws Exception {
+ VectorSchemaRoot r = reader.getVectorSchemaRoot();
+ FieldVector dictA = r.getVector("vectorA");
+ FieldVector dictB = r.getVector("vectorB");
+ FieldVector dictC = r.getVector("vectorC");
+ FieldVector dictD = r.getVector("vectorD");
+
+ switch (state) {
+ case 1:
+ assertDictionary(dictA, reader, valuesPerBlock.get(block)[0]);
+ assertNull(dictB);
+ assertNull(dictC);
+ assertNull(dictD);
+ break;
+ case 2:
+ assertNull(dictA);
+ assertDictionary(dictB, reader, valuesPerBlock.get(block)[1]);
+ assertNull(dictC);
+ assertNull(dictD);
+ break;
+ case 3:
+ assertDictionary(dictA, reader, valuesPerBlock.get(block)[0]);
+ assertDictionary(dictB, reader, valuesPerBlock.get(block)[1]);
+ assertNull(dictC);
+ assertNull(dictD);
+ break;
+ case 4:
+ assertNull(dictA);
+ assertNull(dictB);
+ assertDictionary(dictC, reader, valuesPerBlock.get(block)[2]);
+ assertNull(dictD);
+ break;
+ case 5:
+ assertDictionary(dictA, reader, valuesPerBlock.get(block)[0]);
+ assertNull(dictB);
+ assertDictionary(dictC, reader, valuesPerBlock.get(block)[2]);
+ assertNull(dictD);
+ break;
+ case 6:
+ assertDictionary(dictA, reader, valuesPerBlock.get(block)[0]);
+ assertDictionary(dictB, reader, valuesPerBlock.get(block)[1]);
+ assertDictionary(dictC, reader, valuesPerBlock.get(block)[2]);
+ assertNull(dictD);
+ break;
+ case 7:
+ assertNull(dictA);
+ assertNull(dictB);
+ assertNull(dictC);
+ assertDictionary(dictD, reader, valuesPerBlock.get(block)[3]);
+ break;
+ default:
+ throw new IllegalStateException("Unsupported state: " + state);
+ }
+ }
+
+ protected static Collection<Arguments> dictionaryParams() {
+ List<Arguments> params = new ArrayList<>();
+ for (int i = 1; i < 8; i++) {
Review Comment:
Not clear where these numbers come from.
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
+
+ private int deltaIndex;
+
+ private int dictionaryIndex;
+
+ private boolean wasReset;
+
+ /**
+ * Creates a dictionary with two vectors of the given types. The dictionary vector
+ * will be named "{name}-dictionary".
+ * <p>
+ * To use this dictionary, provide the dictionary vector to a {@link DictionaryProvider},
+ * add the {@link #getIndexVector()} to the {@link org.apache.arrow.vector.VectorSchemaRoot}
+ * and call the {@link #setSafe(int, byte[], int, int)} or other set methods.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator
+ ) {
+ this(name, encoding, dictionaryType, indexType, forFileIPC, allocator, "-dictionary");
+ }
+
+ /**
+ * Creates a dictionary with two vectors of the given types.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ * @param suffix A non-null suffix to append to the name of the dictionary.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator,
+ String suffix
+ ) {
+ this.encoding = encoding;
+ this.forFileIPC = forFileIPC;
+ FieldVector vector = new FieldType(false, dictionaryType, null)
+ .createNewSingleVector(name + suffix, allocator, null);
+ if (!(BaseVariableWidthVector.class.isAssignableFrom(vector.getClass()))) {
+ throw new IllegalArgumentException("Dictionary must be a superclass of 'BaseVariableWidthVector' " +
+ "such as 'VarCharVector'.");
+ }
+ dictionary = (BaseVariableWidthVector) vector;
+ vector = new FieldType(true, indexType, encoding)
+ .createNewSingleVector(name, allocator, null);
+ if (!(BaseIntVector.class.isAssignableFrom(vector.getClass()))) {
+ throw new IllegalArgumentException("Index vector must be a superclass type of 'BaseIntVector' " +
+ "such as 'IntVector' or 'Uint4Vector'.");
+ }
+ indexVector = (BaseIntVector) vector;
+ hashTable = new DictionaryHashTable();
+ }
+
+ /**
+ * Creates a dictionary that will populate the provided vectors with data. Useful if
+ * dictionaries need to be children of a parent vector.
+ * @param dictionary The dictionary to hold the original data.
+ * @param indexVector The index to store the encoded offsets.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ */
+ public BatchedDictionary(
+ FieldVector dictionary,
+ FieldVector indexVector,
+ boolean forFileIPC
+ ) {
+ this.encoding = dictionary.getField().getDictionary();
+ this.forFileIPC = forFileIPC;
+ if (!(BaseVariableWidthVector.class.isAssignableFrom(dictionary.getClass()))) {
+ throw new IllegalArgumentException("Dictionary must be a superclass of 'BaseVariableWidthVector' " +
+ "such as 'VarCharVector'.");
+ }
+ if (dictionary.getField().isNullable()) {
+ throw new IllegalArgumentException("Dictionary must be non-nullable.");
+ }
+ this.dictionary = (BaseVariableWidthVector) dictionary;
+ if (!(BaseIntVector.class.isAssignableFrom(indexVector.getClass()))) {
+ throw new IllegalArgumentException("Index vector must be a superclass type of 'BaseIntVector' " +
+ "such as 'IntVector' or 'Uint4Vector'.");
+ }
+ this.indexVector = (BaseIntVector) indexVector;
+ hashTable = new DictionaryHashTable();
+ }
+
+ /**
+ * The index vector.
+ */
+ public FieldVector getIndexVector() {
+ return indexVector;
+ }
+
+ @Override
+ public FieldVector getVector() {
+ return dictionary;
+ }
+
+ @Override
+ public ArrowType getVectorType() {
+ return dictionary.getField().getType();
+ }
+
+ @Override
+ public DictionaryEncoding getEncoding() {
+ return encoding;
+ }
+
+ /**
+ * Considers the entire byte array as the dictionary value. If the value is null,
+ * a null will be written to the index.
+ *
+ * @param index the value to change
+ * @param value the value to write.
+ */
+ public void setSafe(int index, byte[] value) {
+ if (value == null) {
+ setNull(index);
+ return;
+ }
+ setSafe(index, value, 0, value.length);
+ }
+
+ /**
+ * Encodes the given range in the dictionary. If the value is null, a null will be
+ * written to the index.
+ *
+ * @param index the value to change
+ * @param value the value to write.
+ * @param offset An offset into the value array.
+ * @param len The length of the value to write.
+ */
+ public void setSafe(int index, byte[] value, int offset, int len) {
+ if (value == null || len == 0) {
+ setNull(index);
+ return;
+ }
+ int di = getIndex(value, offset, len);
+ indexVector.setWithPossibleTruncate(index, di);
+ }
+
+ /**
+ * Set the element at the given index to null.
+ *
+ * @param index the value to change
+ */
+ public void setNull(int index) {
+ indexVector.setNull(index);
+ }
+
+ @Override
+ public void close() throws IOException {
+ dictionary.close();
+ indexVector.close();
+ }
+
+ /**
+ * Mark the dictionary as complete for the batch. Called by the {@link ArrowWriter}
+ * on {@link ArrowWriter#writeBatch()}.
+ */
+ public void mark() {
+ dictionary.setValueCount(dictionaryIndex);
+ // not setting the index vector value count. That will happen when the user calls
+ // VectorSchemaRoot#setRowCount().
+ }
+
+ /**
+ * Resets the dictionary to be used for a new batch. Called by the {@link ArrowWriter} on
+ * {@link ArrowWriter#writeBatch()}.
+ */
+ public void reset() {
+ wasReset = true;
Review Comment:
wasReset only ever gets set to true. So if you reset() the dictionary, then populate it again, it is still marked as "wasReset". Is this intentional? It seems like the object basically becomes unusuable if it is reset.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1776046633
@vibhatha ^
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1384285436
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BaseDictionary.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+
+/**
+ * Interface for all dictionary types.
+ */
+public interface BaseDictionary {
Review Comment:
Should we include a bit descriptive comment and explain a bit about the dictionary types (defined here), and relevant details?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "davisusanibar (via GitHub)" <gi...@apache.org>.
davisusanibar commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1419066172
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowReader.java:
##########
@@ -242,6 +247,9 @@ protected void loadDictionary(ArrowDictionaryBatch dictionaryBatch) {
VectorBatchAppender.batchAppend(vector, deltaVector);
}
return;
+ } else if (validateReplacements && getClass() == ArrowFileReader.class) {
+ throw new IllegalStateException("Replacement dictionaries are not supported in " +
+ "the IPC file format. Dictionary ID: " + dictionary.getEncoding().getId());
Review Comment:
> Would it be possible to include these new changes/updates in the IPC documentation file -> docs/source/java/ipc.rst#reading-writing-ipc-formats?
This is not an update or change, it works according to specs, but it will be helpful to present information in a cleaner manner.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1456691820
##########
java/memory/memory-core/src/test/java/org/apache/arrow/memory/util/hash/TestArrowBufHasher.java:
##########
@@ -116,30 +125,36 @@ public void testHasherNegative() {
public void testHasherLessThanInt() {
try (ArrowBuf buf1 = allocator.buffer(4);
ArrowBuf buf2 = allocator.buffer(4)) {
- buf1.writeBytes("foo1".getBytes(StandardCharsets.UTF_8));
- buf2.writeBytes("bar2".getBytes(StandardCharsets.UTF_8));
+ byte[] ba1 = "foo1".getBytes(StandardCharsets.UTF_8);
+ byte[] ba2 = "bar2".getBytes(StandardCharsets.UTF_8);
+ buf1.writeBytes(ba1);
+ buf2.writeBytes(ba2);
for (int i = 1; i <= 4; i ++) {
- verifyHashCodeNotEqual(buf1, 0, i, buf2, 0, i);
+ verifyHashCodeNotEqual(buf1, ba1, 0, i, buf2, ba2, 0, i);
}
}
}
- private void verifyHashCodeNotEqual(ArrowBuf buf1, int offset1, int length1,
- ArrowBuf buf2, int offset2, int length2) {
+ private void verifyHashCodeNotEqual(ArrowBuf buf1, byte[] ba1, int offset1, int length1,
+ ArrowBuf buf2, byte[] ba2, int offset2, int length2) {
int hashCode1 = hasher.hashCode(buf1, 0, length1);
int hashCode2 = hasher.hashCode(buf2, 0, length2);
assertNotEquals(hashCode1, hashCode2);
+
+ hashCode1 = hasher.hashCode(ba1, 0, length1);
+ hashCode2 = hasher.hashCode(ba2, 0, length2);
+ assertNotEquals(hashCode1, hashCode2);
}
@Parameterized.Parameters(name = "hasher = {0}")
public static Collection<Object[]> getHasher() {
return Arrays.asList(
- new Object[] {SimpleHasher.class.getSimpleName(),
- SimpleHasher.INSTANCE},
- new Object[] {MurmurHasher.class.getSimpleName(),
- new MurmurHasher()
- }
+ new Object[] {SimpleHasher.class.getSimpleName(),
+ SimpleHasher.INSTANCE},
+ new Object[] {MurmurHasher.class.getSimpleName(),
+ new MurmurHasher()
+ }
Review Comment:
why this change?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support [arrow]
Posted by "kou (via GitHub)" <gi...@apache.org>.
kou closed pull request #38423: GH-38414 [Java][Vector] Add Delta dictionary support
URL: https://github.com/apache/arrow/pull/38423
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support [arrow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-2501737473
:warning: GitHub issue #38414 **has no components**, please add labels for components.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "davisusanibar (via GitHub)" <gi...@apache.org>.
davisusanibar commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1419208127
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileReader.java:
##########
@@ -164,12 +146,59 @@ public boolean loadNextBatch() throws IOException {
ArrowBlock block = footer.getRecordBatches().get(currentRecordBatch++);
ArrowRecordBatch batch = readRecordBatch(in, block, allocator);
loadRecordBatch(batch);
+ try {
+ loadDictionaries();
+ } catch (IOException e) {
+ batch.close();
+ throw e;
+ }
return true;
} else {
return false;
}
}
+ /**
+ * Loads any dictionaries that may be needed by the given record batch. It attempts
+ * to read as little as possible but may read in more deltas than are necessary for blocks
+ * toward the end of the file.
+ */
+ private void loadDictionaries() throws IOException {
+ // initial load
+ if (currentDictionaryBatch == 0) {
+ for (int i = 0; i < dictionaries.size(); i++) {
+ ArrowBlock block = footer.getDictionaries().get(currentDictionaryBatch++);
+ ArrowDictionaryBatch dictionaryBatch = readDictionaryBatch(in, block, allocator);
+ loadDictionary(dictionaryBatch, false);
+ }
+ estimatedDictionaryRecordBatch++;
+ } else {
+ // we need to look for delta dictionaries. It involves a look-ahead, unfortunately.
+ HashSet<Long> visited = new HashSet<Long>();
+ while (estimatedDictionaryRecordBatch < currentRecordBatch &&
+ currentDictionaryBatch < footer.getDictionaries().size()) {
+ ArrowBlock block = footer.getDictionaries().get(currentDictionaryBatch++);
+ ArrowDictionaryBatch dictionaryBatch = readDictionaryBatch(in, block, allocator);
+ long dictionaryId = dictionaryBatch.getDictionaryId();
+ if (visited.contains(dictionaryId)) {
Review Comment:
Condition 'visited.contains(dictionaryId)' is always 'false'
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385850436
##########
java/vector/src/test/java/org/apache/arrow/vector/ipc/BaseFileTest.java:
##########
@@ -846,4 +862,293 @@ protected void validateListAsMapData(VectorSchemaRoot root) {
}
}
}
+
+ /**
+ * Utility to write permutations of dictionary encoding.
+ *
+ * state == 1, one delta dictionary.
+ * state == 2, one standalone dictionary.
+ * state == 3, one of each
+ * state == 4, delta with nothing at start and end
+ * state == 5, both deltas
+ * state == 6, both deltas and standalone
+ * state == 7, replacement dictionary
+ */
+ protected void writeDataMultiBatchWithDictionaries(OutputStream stream, int state) throws IOException {
+ DictionaryProvider.MapDictionaryProvider provider = new DictionaryProvider.MapDictionaryProvider();
+ DictionaryEncoding deltaEncoding =
+ new DictionaryEncoding(42, false, new ArrowType.Int(16, false), true);
+ DictionaryEncoding replacementEncoding =
+ new DictionaryEncoding(24, false, new ArrowType.Int(16, false), false);
+ DictionaryEncoding deltaCEncoding =
+ new DictionaryEncoding(1, false, new ArrowType.Int(16, false), true);
+ DictionaryEncoding replacementEncodingUpdated =
+ new DictionaryEncoding(2, false, new ArrowType.Int(16, false), false);
+
+ boolean isFile = stream instanceof FileOutputStream;
+ try (BatchedDictionary vectorA = newDictionary("vectorA", deltaEncoding, isFile);
+ BatchedDictionary vectorB = newDictionary("vectorB", replacementEncoding, isFile);
+ BatchedDictionary vectorC = newDictionary("vectorC", deltaCEncoding, isFile);
+ BatchedDictionary vectorD = newDictionary("vectorD", replacementEncodingUpdated, isFile);) {
+ switch (state) {
+ case 1:
+ provider.put(vectorA);
+ break;
+ case 2:
+ provider.put(vectorB);
+ break;
+ case 3:
+ provider.put(vectorA);
+ provider.put(vectorB);
+ break;
+ case 4:
+ provider.put(vectorC);
+ break;
+ case 5:
+ provider.put(vectorA);
+ provider.put(vectorC);
+ break;
+ case 6:
+ provider.put(vectorA);
+ provider.put(vectorB);
+ provider.put(vectorC);
+ break;
+ case 7:
+ provider.put(vectorD);
+ break;
+ default:
+ throw new IllegalStateException("Unsupported state: " + state);
+ }
+
+ VectorSchemaRoot root = null;
+ switch (state) {
+ case 1:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector());
+ break;
+ case 2:
+ root = VectorSchemaRoot.of(vectorB.getIndexVector());
+ break;
+ case 3:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector(), vectorB.getIndexVector());
+ break;
+ case 4:
+ root = VectorSchemaRoot.of(vectorC.getIndexVector());
+ break;
+ case 5:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector(), vectorC.getIndexVector());
+ break;
+ case 6:
+ root = VectorSchemaRoot.of(vectorA.getIndexVector(), vectorB.getIndexVector(), vectorC.getIndexVector());
+ break;
+ case 7:
+ root = VectorSchemaRoot.of(vectorD.getIndexVector());
+ break;
+ default:
+ throw new IllegalStateException("Unsupported state: " + state);
+ }
+
+ ArrowWriter arrowWriter = null;
+ try {
+ if (stream instanceof FileOutputStream) {
+ FileChannel channel = ((FileOutputStream) stream).getChannel();
+ arrowWriter = new ArrowFileWriter(root, provider, channel);
+ } else {
+ arrowWriter = new ArrowStreamWriter(root, provider, stream);
+ }
+
+ vectorA.setSafe(0, "foo".getBytes(StandardCharsets.UTF_8));
+ vectorA.setSafe(1, "bar".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(0, "lorem".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(1, "ipsum".getBytes(StandardCharsets.UTF_8));
+ vectorC.setNull(0);
+ vectorC.setNull(1);
+ vectorD.setSafe(0, "porro".getBytes(StandardCharsets.UTF_8));
+ vectorD.setSafe(1, "amet".getBytes(StandardCharsets.UTF_8));
+
+ // batch 1
+ arrowWriter.start();
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ // batch 2
+ vectorA.setSafe(0, "meep".getBytes(StandardCharsets.UTF_8));
+ vectorA.setSafe(1, "bar".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(0, "ipsum".getBytes(StandardCharsets.UTF_8));
+ vectorB.setSafe(1, "lorem".getBytes(StandardCharsets.UTF_8));
+ vectorC.setSafe(0, "qui".getBytes(StandardCharsets.UTF_8));
+ vectorC.setSafe(1, "dolor".getBytes(StandardCharsets.UTF_8));
+ vectorD.setSafe(0, "amet".getBytes(StandardCharsets.UTF_8));
+ if (state == 7) {
+ vectorD.setSafe(1, "quia".getBytes(StandardCharsets.UTF_8));
+ }
+
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ // batch 3
+ vectorA.setNull(0);
+ vectorA.setNull(1);
+ vectorB.setSafe(0, "ipsum".getBytes(StandardCharsets.UTF_8));
+ vectorB.setNull(1);
+ vectorC.setNull(0);
+ vectorC.setSafe(1, "qui".getBytes(StandardCharsets.UTF_8));
+ vectorD.setNull(0);
+ if (state == 7) {
+ vectorD.setSafe(1, "quia".getBytes(StandardCharsets.UTF_8));
+ }
+
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ // batch 4
+ vectorA.setSafe(0, "bar".getBytes(StandardCharsets.UTF_8));
+ vectorA.setSafe(1, "zap".getBytes(StandardCharsets.UTF_8));
+ vectorB.setNull(0);
+ vectorB.setSafe(1, "lorem".getBytes(StandardCharsets.UTF_8));
+ vectorC.setNull(0);
+ vectorC.setNull(1);
+ if (state == 7) {
+ vectorD.setSafe(0, "quia".getBytes(StandardCharsets.UTF_8));
+ }
+ vectorD.setNull(1);
+
+ root.setRowCount(2);
+ arrowWriter.writeBatch();
+
+ arrowWriter.end();
+ } catch (Exception e) {
+ if (arrowWriter != null) {
+ arrowWriter.close();
+ }
+ throw e;
+ }
+ }
+ }
+
+ Map<Integer, String[][]> valuesPerBlock = new HashMap<Integer, String[][]>();
+
+ {
+ valuesPerBlock.put(0, new String[][]{
+ {"foo", "bar"},
+ {"lorem", "ipsum"},
+ {null, null},
+ {"porro", "amet"}
+ });
+ valuesPerBlock.put(1, new String[][]{
+ {"meep", "bar"},
+ {"ipsum", "lorem"},
+ {"qui", "dolor"},
+ {"amet", "quia"}
+ });
+ valuesPerBlock.put(2, new String[][]{
+ {null, null},
+ {"ipsum", null},
+ {null, "qui"},
+ {null, "quia"}
+ });
+ valuesPerBlock.put(3, new String[][]{
+ {"bar", "zap"},
+ {null, "lorem"},
+ {null, null},
+ {"quia", null}
+ });
+ }
+
+ protected void assertDictionary(FieldVector encoded, ArrowReader reader, String... expected) throws Exception {
+ DictionaryEncoding dictionaryEncoding = encoded.getField().getDictionary();
+ BaseDictionary dictionary = reader.getDictionaryVectors().get(dictionaryEncoding.getId());
+ try (ValueVector decoded = DictionaryEncoder.decode(encoded, dictionary)) {
+ Assertions.assertEquals(expected.length, encoded.getValueCount());
+ for (int i = 0; i < expected.length; i++) {
+ if (expected[i] == null) {
+ Assertions.assertNull(decoded.getObject(i));
Review Comment:
ack
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385882048
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BaseDictionary.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+
+/**
+ * Interface for all dictionary types.
+ */
+public interface BaseDictionary {
Review Comment:
Ack
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385845189
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
+
+ private int deltaIndex;
+
+ private int dictionaryIndex;
+
+ private boolean wasReset;
+
+ /**
+ * Creates a dictionary with two vectors of the given types. The dictionary vector
+ * will be named "{name}-dictionary".
+ * <p>
+ * To use this dictionary, provide the dictionary vector to a {@link DictionaryProvider},
+ * add the {@link #getIndexVector()} to the {@link org.apache.arrow.vector.VectorSchemaRoot}
+ * and call the {@link #setSafe(int, byte[], int, int)} or other set methods.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator
+ ) {
+ this(name, encoding, dictionaryType, indexType, forFileIPC, allocator, "-dictionary");
+ }
+
+ /**
+ * Creates a dictionary with two vectors of the given types.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ * @param suffix A non-null suffix to append to the name of the dictionary.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator,
+ String suffix
+ ) {
+ this.encoding = encoding;
+ this.forFileIPC = forFileIPC;
+ FieldVector vector = new FieldType(false, dictionaryType, null)
+ .createNewSingleVector(name + suffix, allocator, null);
+ if (!(BaseVariableWidthVector.class.isAssignableFrom(vector.getClass()))) {
+ throw new IllegalArgumentException("Dictionary must be a superclass of 'BaseVariableWidthVector' " +
+ "such as 'VarCharVector'.");
+ }
+ dictionary = (BaseVariableWidthVector) vector;
+ vector = new FieldType(true, indexType, encoding)
+ .createNewSingleVector(name, allocator, null);
+ if (!(BaseIntVector.class.isAssignableFrom(vector.getClass()))) {
+ throw new IllegalArgumentException("Index vector must be a superclass type of 'BaseIntVector' " +
+ "such as 'IntVector' or 'Uint4Vector'.");
+ }
+ indexVector = (BaseIntVector) vector;
+ hashTable = new DictionaryHashTable();
+ }
+
+ /**
+ * Creates a dictionary that will populate the provided vectors with data. Useful if
+ * dictionaries need to be children of a parent vector.
+ * @param dictionary The dictionary to hold the original data.
+ * @param indexVector The index to store the encoded offsets.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ */
+ public BatchedDictionary(
+ FieldVector dictionary,
+ FieldVector indexVector,
+ boolean forFileIPC
+ ) {
+ this.encoding = dictionary.getField().getDictionary();
+ this.forFileIPC = forFileIPC;
+ if (!(BaseVariableWidthVector.class.isAssignableFrom(dictionary.getClass()))) {
+ throw new IllegalArgumentException("Dictionary must be a superclass of 'BaseVariableWidthVector' " +
+ "such as 'VarCharVector'.");
+ }
+ if (dictionary.getField().isNullable()) {
+ throw new IllegalArgumentException("Dictionary must be non-nullable.");
+ }
+ this.dictionary = (BaseVariableWidthVector) dictionary;
+ if (!(BaseIntVector.class.isAssignableFrom(indexVector.getClass()))) {
+ throw new IllegalArgumentException("Index vector must be a superclass type of 'BaseIntVector' " +
+ "such as 'IntVector' or 'Uint4Vector'.");
+ }
+ this.indexVector = (BaseIntVector) indexVector;
+ hashTable = new DictionaryHashTable();
+ }
+
+ /**
+ * The index vector.
+ */
+ public FieldVector getIndexVector() {
+ return indexVector;
+ }
+
+ @Override
+ public FieldVector getVector() {
+ return dictionary;
+ }
+
+ @Override
+ public ArrowType getVectorType() {
+ return dictionary.getField().getType();
+ }
+
+ @Override
+ public DictionaryEncoding getEncoding() {
+ return encoding;
+ }
+
+ /**
+ * Considers the entire byte array as the dictionary value. If the value is null,
+ * a null will be written to the index.
+ *
+ * @param index the value to change
+ * @param value the value to write.
+ */
+ public void setSafe(int index, byte[] value) {
+ if (value == null) {
+ setNull(index);
+ return;
+ }
+ setSafe(index, value, 0, value.length);
+ }
+
+ /**
+ * Encodes the given range in the dictionary. If the value is null, a null will be
+ * written to the index.
+ *
+ * @param index the value to change
+ * @param value the value to write.
+ * @param offset An offset into the value array.
+ * @param len The length of the value to write.
+ */
+ public void setSafe(int index, byte[] value, int offset, int len) {
+ if (value == null || len == 0) {
+ setNull(index);
+ return;
+ }
+ int di = getIndex(value, offset, len);
+ indexVector.setWithPossibleTruncate(index, di);
+ }
+
+ /**
+ * Set the element at the given index to null.
+ *
+ * @param index the value to change
+ */
+ public void setNull(int index) {
+ indexVector.setNull(index);
+ }
+
+ @Override
+ public void close() throws IOException {
+ dictionary.close();
+ indexVector.close();
+ }
+
+ /**
+ * Mark the dictionary as complete for the batch. Called by the {@link ArrowWriter}
+ * on {@link ArrowWriter#writeBatch()}.
+ */
+ public void mark() {
+ dictionary.setValueCount(dictionaryIndex);
+ // not setting the index vector value count. That will happen when the user calls
+ // VectorSchemaRoot#setRowCount().
+ }
+
+ /**
+ * Resets the dictionary to be used for a new batch. Called by the {@link ArrowWriter} on
+ * {@link ArrowWriter#writeBatch()}.
+ */
+ public void reset() {
+ wasReset = true;
Review Comment:
It's intended as a fail-fast flag so that if the user populates the dictionary, writes a batch to a file, then tries to write a replacement dictionary, the exception would be thrown before the next call to `writeBatch`. Part of the reason I left the TODO's in there.
If ya'll (and @danepitkin) think it's ok to wait until a call to `writeBatch` then I can drop that `forFileIPC` flag and catch the error in the `ArrowFileWriter`. I'd still need a flag like the reset flag somewhere to track if a dictionary was a replacement or not.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1457461064
##########
java/vector/src/test/java/org/apache/arrow/vector/ipc/TestArrowStream.java:
##########
@@ -64,7 +82,7 @@ public void testStreamZeroLengthBatch() throws IOException {
try (IntVector vector = new IntVector("foo", allocator);) {
Schema schema = new Schema(Collections.singletonList(vector.getField()));
try (VectorSchemaRoot root =
- new VectorSchemaRoot(schema, Collections.singletonList(vector), vector.getValueCount());
+ new VectorSchemaRoot(schema, Collections.singletonList(vector), vector.getValueCount());
Review Comment:
CC @danepitkin @davisusanibar I think it would help contributors if we used a formatter like google-java-format instead of peppering them with questions like this that checkstyle apparently doesn't handle
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385270989
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowReader.java:
##########
@@ -243,6 +248,8 @@ protected void loadDictionary(ArrowDictionaryBatch dictionaryBatch) {
}
return;
}
+ // TODO - If the super class is a file reader it may be good to throw an exception here
+ // the dictionary failed to satisfy the spec (i.e. being a replacement dictionary)
Review Comment:
Ack, do you leave TODOs in or just remove them and create the issue?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1777548326
Hi! Yes, still working through the JNI bits.
Regarding #38169, this is just the Delta encoding and doesn't address changing the format. I'll work on that later with the community.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1776037229
:warning: GitHub issue #38414 **has been automatically assigned in GitHub** to PR creator.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1777474247
Seems like the previous error is fixed now, but the same test is failing in most of the CIs.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385840863
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
+
+ private int deltaIndex;
+
+ private int dictionaryIndex;
+
+ private boolean wasReset;
+
+ /**
+ * Creates a dictionary with two vectors of the given types. The dictionary vector
+ * will be named "{name}-dictionary".
+ * <p>
+ * To use this dictionary, provide the dictionary vector to a {@link DictionaryProvider},
+ * add the {@link #getIndexVector()} to the {@link org.apache.arrow.vector.VectorSchemaRoot}
+ * and call the {@link #setSafe(int, byte[], int, int)} or other set methods.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator
+ ) {
+ this(name, encoding, dictionaryType, indexType, forFileIPC, allocator, "-dictionary");
+ }
+
+ /**
+ * Creates a dictionary with two vectors of the given types.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ * @param suffix A non-null suffix to append to the name of the dictionary.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator,
+ String suffix
+ ) {
+ this.encoding = encoding;
+ this.forFileIPC = forFileIPC;
+ FieldVector vector = new FieldType(false, dictionaryType, null)
+ .createNewSingleVector(name + suffix, allocator, null);
+ if (!(BaseVariableWidthVector.class.isAssignableFrom(vector.getClass()))) {
+ throw new IllegalArgumentException("Dictionary must be a superclass of 'BaseVariableWidthVector' " +
+ "such as 'VarCharVector'.");
+ }
+ dictionary = (BaseVariableWidthVector) vector;
+ vector = new FieldType(true, indexType, encoding)
+ .createNewSingleVector(name, allocator, null);
+ if (!(BaseIntVector.class.isAssignableFrom(vector.getClass()))) {
+ throw new IllegalArgumentException("Index vector must be a superclass type of 'BaseIntVector' " +
Review Comment:
ah good catch.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1385849993
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileReader.java:
##########
@@ -164,12 +146,54 @@ public boolean loadNextBatch() throws IOException {
ArrowBlock block = footer.getRecordBatches().get(currentRecordBatch++);
ArrowRecordBatch batch = readRecordBatch(in, block, allocator);
loadRecordBatch(batch);
+ loadDictionaries();
Review Comment:
Yup, will fix.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on PR #38423:
URL: https://github.com/apache/arrow/pull/38423#issuecomment-1791153538
@vibhatha @jbonofre poke :)
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "davisusanibar (via GitHub)" <gi...@apache.org>.
davisusanibar commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1419830287
##########
java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileReader.java:
##########
@@ -164,12 +146,59 @@ public boolean loadNextBatch() throws IOException {
ArrowBlock block = footer.getRecordBatches().get(currentRecordBatch++);
ArrowRecordBatch batch = readRecordBatch(in, block, allocator);
loadRecordBatch(batch);
+ try {
+ loadDictionaries();
+ } catch (IOException e) {
+ batch.close();
+ throw e;
+ }
return true;
} else {
return false;
}
}
+ /**
+ * Loads any dictionaries that may be needed by the given record batch. It attempts
+ * to read as little as possible but may read in more deltas than are necessary for blocks
+ * toward the end of the file.
+ */
+ private void loadDictionaries() throws IOException {
+ // initial load
+ if (currentDictionaryBatch == 0) {
+ for (int i = 0; i < dictionaries.size(); i++) {
+ ArrowBlock block = footer.getDictionaries().get(currentDictionaryBatch++);
+ ArrowDictionaryBatch dictionaryBatch = readDictionaryBatch(in, block, allocator);
+ loadDictionary(dictionaryBatch, false);
+ }
+ estimatedDictionaryRecordBatch++;
+ } else {
+ // we need to look for delta dictionaries. It involves a look-ahead, unfortunately.
+ HashSet<Long> visited = new HashSet<Long>();
+ while (estimatedDictionaryRecordBatch < currentRecordBatch &&
+ currentDictionaryBatch < footer.getDictionaries().size()) {
+ ArrowBlock block = footer.getDictionaries().get(currentDictionaryBatch++);
+ ArrowDictionaryBatch dictionaryBatch = readDictionaryBatch(in, block, allocator);
+ long dictionaryId = dictionaryBatch.getDictionaryId();
+ if (visited.contains(dictionaryId)) {
+ // done
+ currentDictionaryBatch--;
+ estimatedDictionaryRecordBatch++;
+ } else if (!dictionaries.containsKey(dictionaryId)) {
+ throw new IOException("Dictionary ID " + dictionaryId + " was written " +
+ "after the initial batch. The file does not follow the IPC file protocol.");
+ } else if (!dictionaryBatch.isDelta()) {
+ throw new IOException("Dictionary ID " + dictionaryId + " was written as a replacement " +
+ "after the initial batch. Replacement dictionaries are not currently allowed in the IPC file protocol.");
+ } else {
+ loadDictionary(dictionaryBatch, true);
+ }
+ }
+ if (currentDictionaryBatch >= footer.getDictionaries().size()) {
Review Comment:
Is it necessary? Would it be possible for you to validate if the unit test works with or without this validation
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1457769910
##########
java/vector/src/test/java/org/apache/arrow/vector/ipc/TestArrowStream.java:
##########
@@ -64,7 +82,7 @@ public void testStreamZeroLengthBatch() throws IOException {
try (IntVector vector = new IntVector("foo", allocator);) {
Schema schema = new Schema(Collections.singletonList(vector.getField()));
try (VectorSchemaRoot root =
- new VectorSchemaRoot(schema, Collections.singletonList(vector), vector.getValueCount());
+ new VectorSchemaRoot(schema, Collections.singletonList(vector), vector.getValueCount());
Review Comment:
No, I'll clean up all of these silly formatting issues.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1457820246
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation for continuous encoding of data in a dictionary and
+ * index vector as opposed to the {@link Dictionary} that encodes a complete vector.
+ * Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private int deltaIndex;
+
+ private int dictionaryIndex;
+
+ private boolean oneTimeEncoding;
+
+ private boolean wasReset;
+
+ private boolean vectorsProvided;
+
+ /**
+ * Creates a dictionary and index vector with the respective types. The dictionary vector
+ * will be named "{name}-dictionary".
+ * <p>
+ * To use this dictionary, provide the dictionary vector to a {@link DictionaryProvider},
+ * add the {@link #getIndexVector()} to the {@link org.apache.arrow.vector.VectorSchemaRoot}
+ * and call the {@link #setSafe(int, byte[], int, int)} or other set methods.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param allocator The allocator to use.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ BufferAllocator allocator
+ ) {
+ this(name, encoding, dictionaryType, indexType, allocator, "-dictionary");
+ }
+
+ /**
+ * Creates a dictionary index vector with the respective types.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param allocator The allocator to use.
+ * @param suffix A non-null suffix to append to the name of the dictionary.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ BufferAllocator allocator,
+ String suffix
+ ) {
+ this(name, encoding, dictionaryType, indexType, allocator, suffix, false);
+ }
+
+ /**
+ * Creates a dictionary index vector with the respective types.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param allocator The allocator to use.
+ * @param suffix A non-null suffix to append to the name of the dictionary.
+ * @param oneTimeEncoding A mode where the entries can be added to the dictionary until
+ * the first stream batch is written. After that, any new entries
+ * to the dictionary will throw an exception.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ BufferAllocator allocator,
+ String suffix,
+ boolean oneTimeEncoding
+ ) {
+ this.encoding = encoding;
+ this.oneTimeEncoding = oneTimeEncoding;
+ if (dictionaryType.getTypeID() != ArrowType.ArrowTypeID.Utf8 &&
+ dictionaryType.getTypeID() != ArrowType.ArrowTypeID.Binary) {
Review Comment:
No, of course, but I was limiting this for string and binary data. Now that I've seen more examples, it looks like folks may use dicts for larger integers or floats as well so a question is whether or not to extend this class for those types or maybe rename this as `BaseBinaryDictionary` to work with only bytes and strings. Thoughts?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1459547277
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ipc.ArrowWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation that can be used when writing batches of data to
+ * a stream or file. Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private final boolean forFileIPC;
+
+ private int deltaIndex;
+
+ private int dictionaryIndex;
+
+ private boolean wasReset;
+
+ /**
+ * Creates a dictionary with two vectors of the given types. The dictionary vector
+ * will be named "{name}-dictionary".
+ * <p>
+ * To use this dictionary, provide the dictionary vector to a {@link DictionaryProvider},
+ * add the {@link #getIndexVector()} to the {@link org.apache.arrow.vector.VectorSchemaRoot}
+ * and call the {@link #setSafe(int, byte[], int, int)} or other set methods.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator
+ ) {
+ this(name, encoding, dictionaryType, indexType, forFileIPC, allocator, "-dictionary");
+ }
+
+ /**
+ * Creates a dictionary with two vectors of the given types.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ * @param allocator The allocator to use.
+ * @param suffix A non-null suffix to append to the name of the dictionary.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ boolean forFileIPC,
+ BufferAllocator allocator,
+ String suffix
+ ) {
+ this.encoding = encoding;
+ this.forFileIPC = forFileIPC;
+ FieldVector vector = new FieldType(false, dictionaryType, null)
+ .createNewSingleVector(name + suffix, allocator, null);
+ if (!(BaseVariableWidthVector.class.isAssignableFrom(vector.getClass()))) {
+ throw new IllegalArgumentException("Dictionary must be a superclass of 'BaseVariableWidthVector' " +
+ "such as 'VarCharVector'.");
+ }
+ dictionary = (BaseVariableWidthVector) vector;
+ vector = new FieldType(true, indexType, encoding)
+ .createNewSingleVector(name, allocator, null);
+ if (!(BaseIntVector.class.isAssignableFrom(vector.getClass()))) {
+ throw new IllegalArgumentException("Index vector must be a superclass type of 'BaseIntVector' " +
+ "such as 'IntVector' or 'Uint4Vector'.");
+ }
+ indexVector = (BaseIntVector) vector;
+ hashTable = new DictionaryHashTable();
+ }
+
+ /**
+ * Creates a dictionary that will populate the provided vectors with data. Useful if
+ * dictionaries need to be children of a parent vector.
+ * @param dictionary The dictionary to hold the original data.
+ * @param indexVector The index to store the encoded offsets.
+ * @param forFileIPC Whether the data will be written to a file or stream IPC. Throws an
+ * exception if a replacement dictionary is provided to a file IPC.
+ */
+ public BatchedDictionary(
+ FieldVector dictionary,
+ FieldVector indexVector,
+ boolean forFileIPC
+ ) {
+ this.encoding = dictionary.getField().getDictionary();
+ this.forFileIPC = forFileIPC;
+ if (!(BaseVariableWidthVector.class.isAssignableFrom(dictionary.getClass()))) {
+ throw new IllegalArgumentException("Dictionary must be a superclass of 'BaseVariableWidthVector' " +
+ "such as 'VarCharVector'.");
+ }
+ if (dictionary.getField().isNullable()) {
+ throw new IllegalArgumentException("Dictionary must be non-nullable.");
+ }
+ this.dictionary = (BaseVariableWidthVector) dictionary;
+ if (!(BaseIntVector.class.isAssignableFrom(indexVector.getClass()))) {
+ throw new IllegalArgumentException("Index vector must be a superclass type of 'BaseIntVector' " +
+ "such as 'IntVector' or 'Uint4Vector'.");
+ }
+ this.indexVector = (BaseIntVector) indexVector;
+ hashTable = new DictionaryHashTable();
+ }
+
+ /**
+ * The index vector.
+ */
+ public FieldVector getIndexVector() {
+ return indexVector;
+ }
+
+ @Override
+ public FieldVector getVector() {
+ return dictionary;
+ }
+
+ @Override
+ public ArrowType getVectorType() {
+ return dictionary.getField().getType();
+ }
+
+ @Override
+ public DictionaryEncoding getEncoding() {
+ return encoding;
+ }
+
+ /**
+ * Considers the entire byte array as the dictionary value. If the value is null,
+ * a null will be written to the index.
+ *
+ * @param index the value to change
+ * @param value the value to write.
+ */
+ public void setSafe(int index, byte[] value) {
+ if (value == null) {
+ setNull(index);
+ return;
+ }
+ setSafe(index, value, 0, value.length);
+ }
+
+ /**
+ * Encodes the given range in the dictionary. If the value is null, a null will be
+ * written to the index.
+ *
+ * @param index the value to change
+ * @param value the value to write.
+ * @param offset An offset into the value array.
+ * @param len The length of the value to write.
+ */
+ public void setSafe(int index, byte[] value, int offset, int len) {
+ if (value == null || len == 0) {
+ setNull(index);
+ return;
+ }
+ int di = getIndex(value, offset, len);
+ indexVector.setWithPossibleTruncate(index, di);
+ }
+
+ /**
+ * Set the element at the given index to null.
+ *
+ * @param index the value to change
+ */
+ public void setNull(int index) {
+ indexVector.setNull(index);
+ }
+
+ @Override
+ public void close() throws IOException {
+ dictionary.close();
+ indexVector.close();
+ }
+
+ /**
+ * Mark the dictionary as complete for the batch. Called by the {@link ArrowWriter}
+ * on {@link ArrowWriter#writeBatch()}.
+ */
+ public void mark() {
+ dictionary.setValueCount(dictionaryIndex);
+ // not setting the index vector value count. That will happen when the user calls
+ // VectorSchemaRoot#setRowCount().
+ }
+
+ /**
+ * Resets the dictionary to be used for a new batch. Called by the {@link ArrowWriter} on
+ * {@link ArrowWriter#writeBatch()}.
+ */
+ public void reset() {
+ wasReset = true;
Review Comment:
In this case it is used to reset the state, depending on delta (maintain memo but reset the dictionary vector so it can hold new unique values) or replacement (reset dictionary _and_ memo) _and_ stream or file mode (throw error on update before flushing a batch). Definitely confusing. I'd rather update the file spec to support replacements.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "manolama (via GitHub)" <gi...@apache.org>.
manolama commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1459539156
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/Dictionary.java:
##########
@@ -72,4 +74,14 @@ public boolean equals(Object o) {
public int hashCode() {
return Objects.hash(encoding, dictionary);
}
+
+ @Override
+ public int mark() {
+ return reset == true ? 0 : dictionary.getValueCount();
+ }
+
+ @Override
+ public void reset() {
+ reset = true;
+ }
Review Comment:
Yeah, that's fair, the state is ugly. I'm trying to keep the API changes to a minimum for backwards compatibility and make it easy on devs to write multiple batches in the current idiom.
Taking CPP as the canonical implementation, their writer has a method to write replacement or delta dictionaries before a batch but the Java API assumes immutable dictionaries are provided at writer creation via the `DictionaryProvider`.
We _could_ add an `DictionaryProvider.update(Dictionary dictionary)` method that mimics the CPP behavior (and have a dictionary builder helper like `builder_dict`) but I'd still need a bit of state in the provider to determine when a batch write was performed. Without that state, the only way I know of to determine if a dictionary has changed and should be flushed in a batch is to hash on the _contents_ of the dictionary vector on every batch write. That would really hurt performance with large dictionaries.
What do you 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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java][Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1458200405
##########
java/vector/src/test/java/org/apache/arrow/vector/ipc/TestArrowStream.java:
##########
@@ -64,7 +82,7 @@ public void testStreamZeroLengthBatch() throws IOException {
try (IntVector vector = new IntVector("foo", allocator);) {
Schema schema = new Schema(Collections.singletonList(vector.getField()));
try (VectorSchemaRoot root =
- new VectorSchemaRoot(schema, Collections.singletonList(vector), vector.getValueCount());
+ new VectorSchemaRoot(schema, Collections.singletonList(vector), vector.getValueCount());
Review Comment:
Sorry about the repeated nits on the formatting. I agree with @lidavidm's idea.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38414 [Java] [Vector] Add Delta dictionary support. [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #38423:
URL: https://github.com/apache/arrow/pull/38423#discussion_r1424816799
##########
java/vector/src/main/java/org/apache/arrow/vector/dictionary/BatchedDictionary.java:
##########
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.dictionary;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.hash.MurmurHasher;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.arrow.vector.BaseIntVector;
+import org.apache.arrow.vector.BaseVariableWidthVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/**
+ * A dictionary implementation for continuous encoding of data in a dictionary and
+ * index vector as opposed to the {@link Dictionary} that encodes a complete vector.
+ * Supports delta or replacement encoding.
+ */
+public class BatchedDictionary implements Closeable, BaseDictionary {
+
+ private final DictionaryEncoding encoding;
+
+ private final BaseVariableWidthVector dictionary;
+
+ private final BaseIntVector indexVector;
+
+ private final DictionaryHashTable hashTable;
+
+ private int deltaIndex;
+
+ private int dictionaryIndex;
+
+ private boolean oneTimeEncoding;
+
+ private boolean wasReset;
+
+ private boolean vectorsProvided;
+
+ /**
+ * Creates a dictionary and index vector with the respective types. The dictionary vector
+ * will be named "{name}-dictionary".
+ * <p>
+ * To use this dictionary, provide the dictionary vector to a {@link DictionaryProvider},
+ * add the {@link #getIndexVector()} to the {@link org.apache.arrow.vector.VectorSchemaRoot}
+ * and call the {@link #setSafe(int, byte[], int, int)} or other set methods.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param allocator The allocator to use.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ BufferAllocator allocator
+ ) {
+ this(name, encoding, dictionaryType, indexType, allocator, "-dictionary");
+ }
+
+ /**
+ * Creates a dictionary index vector with the respective types.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param allocator The allocator to use.
+ * @param suffix A non-null suffix to append to the name of the dictionary.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ BufferAllocator allocator,
+ String suffix
+ ) {
+ this(name, encoding, dictionaryType, indexType, allocator, suffix, false);
+ }
+
+ /**
+ * Creates a dictionary index vector with the respective types.
+ *
+ * @param name A name for the vector and dictionary.
+ * @param encoding The dictionary encoding to use.
+ * @param dictionaryType The type of the dictionary data.
+ * @param indexType The type of the encoded dictionary index.
+ * @param allocator The allocator to use.
+ * @param suffix A non-null suffix to append to the name of the dictionary.
+ * @param oneTimeEncoding A mode where the entries can be added to the dictionary until
+ * the first stream batch is written. After that, any new entries
+ * to the dictionary will throw an exception.
+ */
+ public BatchedDictionary(
+ String name,
+ DictionaryEncoding encoding,
+ ArrowType dictionaryType,
+ ArrowType indexType,
+ BufferAllocator allocator,
+ String suffix,
+ boolean oneTimeEncoding
+ ) {
+ this.encoding = encoding;
+ this.oneTimeEncoding = oneTimeEncoding;
+ if (dictionaryType.getTypeID() != ArrowType.ArrowTypeID.Utf8 &&
+ dictionaryType.getTypeID() != ArrowType.ArrowTypeID.Binary) {
Review Comment:
Can the `dictionaryType.getTypeID()` be `Utf8` and `Binary` at the same time?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org