This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 04a294f3e2e Add BSON (MongoDB) input format support (#18950)
04a294f3e2e is described below
commit 04a294f3e2e393609e256efa8e2b95e2e20f5bfc
Author: Xiang Fu <[email protected]>
AuthorDate: Fri Jul 10 07:50:58 2026 +0200
Add BSON (MongoDB) input format support (#18950)
* Add BSON (MongoDB) input format support
Add a new `pinot-bson` input-format plugin for reading MongoDB BSON data,
covering both batch and streaming ingestion:
- BSONRecordReader reads mongodump-style files (concatenated length-prefixed
BSON documents); gzip files are supported transparently.
- BSONMessageDecoder decodes one BSON document per stream message.
- BSONRecordExtractor maps BSON's types onto the RecordExtractor contract:
ObjectId -> hex String, DateTime -> java.sql.Timestamp, Decimal128 ->
BigDecimal (NaN/Infinity -> null), Binary -> byte[], embedded documents ->
Map, arrays -> Object[], with a toString() fallback for rare/deprecated
types.
BSON is registered in the FileFormat enum and RecordReaderFactory so
`dataFormat: bson` works in batch ingestion out of the box, and the module
is
wired into the BOM, distribution assembly, tools, and LICENSE-binary. Uses
the
standalone org.mongodb:bson library (no MongoDB driver / network
dependency).
* Address PR review comments
- BSONRecordReader: on a truncated/corrupt tail, emit the already-read valid
record and defer the fetch error to the following next() call (via a
stashed
IOException surfaced through hasNext()/next()), instead of discarding the
last
valid record. Matches the emit-then-error behavior of JSONRecordReader.
- BSONMessageDecoderTest: use assertSame instead of assertTrue(result ==
row).
- Add a reader regression test covering the truncated-tail case.
* Address review: bound frame length, handle negative-zero Decimal128
- BSONRecordReader: reject a frame length above the 16MB BSON maximum. A
corrupt
length prefix (e.g. 0x7FFFFFFF) previously reached `new byte[length]` and
raised
OutOfMemoryError, an Error that escapes the
catch(RuntimeException)/catch(IOException)
recovery paths and aborts the segment build. It now surfaces as a
recoverable
IOException before any allocation.
- BSONRecordExtractor: negative-zero Decimal128 (a legal Mongo value, at
any exponent)
has isNaN()==false and isInfinite()==false, so it slipped the guard and
threw
ArithmeticException out of extract(). It now converts to BigDecimal.ZERO.
There is no
public negative-zero predicate and Decimal128.parse("-0.00") !=
NEGATIVE_ZERO, so the
exception is caught rather than pre-checked.
- BSONRecordReader.open(): close the input stream if the first read fails,
matching
JSONRecordReader.init().
- BSONUtils: hoist the immutable DecoderContext to a static final field.
- BSONRecordExtractor javadoc: correct the toString() fallback wording --
it also covers
the internal Timestamp type and the 5.x binary vector types.
- Add regression tests for both fixed edge cases.
* Address review: map BsonTimestamp, pin fallback strings, cover bson
registration
- BsonTimestamp (the oplog `ts` / change-stream `clusterTime` field)
previously
fell through to value.toString() and ingested as the debug string
"Timestamp{value=..., seconds=..., inc=...}". Convert it to
java.sql.Timestamp
at second granularity, matching MongoDB's $toDate. The intra-second
ordinal
counter is not representable and is dropped.
- The remaining toString() fallbacks (MinKey, MaxKey, Symbol, Code,
BsonRegularExpression) render via formats that org.mongodb:bson does not
document, yet they are baked into segment data.
testExoticTypeFallsBackToToString
asserted extract(x).equals(x.toString()), which is self-referential and
pins
nothing. Pin the literal strings so a driver upgrade that reformats them
fails
the build instead of silently rewriting segments.
- Add a round-trip test for BSON binary subtypes 0x03/0x04 (how MongoDB
stores
UUIDs). They currently decode to Binary -> byte[] because DocumentCodec
defaults
to UuidRepresentation.UNSPECIFIED; pin that so a default flip to
java.util.UUID
(which would change the column from BYTES to a 36-char STRING) is caught.
- RecordReaderFactoryTest enumerates every format's default reader class but
omitted bson, leaving `dataFormat: bson` resolution unverified and a typo
in
DEFAULT_BSON_RECORD_READER_CLASS a runtime-only failure.
- Document BSONUtils' thread safety (the shared
DocumentCodec/DecoderContext are
immutable, so concurrent ingestion threads may decode). Note that
Decimal128
extends Number, so its branch must stay above the Number pass-through.
- Use /// Javadoc consistently across the plugin.
* Read the BsonTimestamp seconds field as unsigned
BSON stores a Timestamp's seconds component as an unsigned 32-bit field, but
BsonTimestamp#getTime() returns it as a signed int. From
2038-01-19T03:14:08Z
onward the high bit is set, so widening the raw value sign-extended it: a
2039-01-01 oplog timestamp ingested as 1902-11-25.
Mask before widening. This matches the MongoDB server, which reads those
seconds
as unsigned.
---
AGENTS.md | 1 +
CLAUDE.md | 1 +
LICENSE-binary | 1 +
pinot-bom/pom.xml | 5 +
pinot-distribution/pinot-assembly.xml | 6 +
.../pinot-input-format/{ => pinot-bson}/pom.xml | 42 ++-
.../inputformat/bson/BSONMessageDecoder.java | 67 +++++
.../inputformat/bson/BSONRecordExtractor.java | 147 +++++++++++
.../plugin/inputformat/bson/BSONRecordReader.java | 168 ++++++++++++
.../pinot/plugin/inputformat/bson/BSONUtils.java | 57 ++++
.../inputformat/bson/BSONMessageDecoderTest.java | 153 +++++++++++
.../inputformat/bson/BSONRecordExtractorTest.java | 290 +++++++++++++++++++++
.../inputformat/bson/BSONRecordReaderTest.java | 212 +++++++++++++++
.../pinot-bson/src/test/resources/log4j2.xml | 36 +++
pinot-plugins/pinot-input-format/pom.xml | 1 +
.../apache/pinot/spi/data/readers/FileFormat.java | 2 +-
.../spi/data/readers/RecordReaderFactory.java | 2 +
.../spi/data/readers/RecordReaderFactoryTest.java | 4 +
pinot-tools/pom.xml | 4 +
pom.xml | 6 +
20 files changed, 1178 insertions(+), 27 deletions(-)
diff --git a/AGENTS.md b/AGENTS.md
index eec4f372c2d..cd62355a0d7 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -48,6 +48,7 @@ repo. It is intentionally short and focused on day-to-day
work.
- pinot-arrow: Apache Arrow input format support.
- pinot-avro: Avro input format support.
- pinot-avro-base: shared Avro utilities and base classes.
+ - pinot-bson: MongoDB BSON input format support.
- pinot-clp-log: CLP log input format support.
- pinot-confluent-avro: Confluent Schema Registry Avro input support.
- pinot-confluent-json: Confluent Schema Registry JSON input support.
diff --git a/CLAUDE.md b/CLAUDE.md
index 25cd4c2de28..d9abfc33e77 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -44,6 +44,7 @@ Apache Pinot is a real-time distributed OLAP datastore for
low-latency analytics
- `pinot-arrow`: Apache Arrow input format support.
- `pinot-avro`: Avro input format support.
- `pinot-avro-base`: shared Avro utilities and base classes.
+ - `pinot-bson`: MongoDB BSON input format support.
- `pinot-clp-log`: CLP log input format support.
- `pinot-confluent-avro`: Confluent Schema Registry Avro input support.
- `pinot-confluent-json`: Confluent Schema Registry JSON input support.
diff --git a/LICENSE-binary b/LICENSE-binary
index b7c9486ba7a..8d688bb0936 100644
--- a/LICENSE-binary
+++ b/LICENSE-binary
@@ -531,6 +531,7 @@ org.jooq:joou-java-6:0.9.5
org.jspecify:jspecify:1.0.0
org.locationtech.proj4j:proj4j:1.2.2
org.lz4:lz4-java:1.8.0
+org.mongodb:bson:5.6.1
org.quartz-scheduler:quartz:2.5.2
org.roaringbitmap:RoaringBitmap:1.6.12
org.scala-lang.modules:scala-collection-compat_2.13:2.10.0
diff --git a/pinot-bom/pom.xml b/pinot-bom/pom.xml
index aa67b76074b..4d8566f29c0 100644
--- a/pinot-bom/pom.xml
+++ b/pinot-bom/pom.xml
@@ -324,6 +324,11 @@
<artifactId>pinot-json</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-bson</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-csv</artifactId>
diff --git a/pinot-distribution/pinot-assembly.xml
b/pinot-distribution/pinot-assembly.xml
index 80747081d3e..13940e20ca3 100644
--- a/pinot-distribution/pinot-assembly.xml
+++ b/pinot-distribution/pinot-assembly.xml
@@ -152,6 +152,12 @@
</source>
<destName>plugins/pinot-input-format/pinot-json/pinot-json-${project.version}-shaded.jar</destName>
</file>
+ <file>
+ <source>
+
${pinot.root}/pinot-plugins/pinot-input-format/pinot-bson/target/pinot-bson-${project.version}-shaded.jar
+ </source>
+
<destName>plugins/pinot-input-format/pinot-bson/pinot-bson-${project.version}-shaded.jar</destName>
+ </file>
<file>
<source>${pinot.root}/pinot-plugins/pinot-input-format/pinot-orc/target/pinot-orc-${project.version}-shaded.jar
</source>
diff --git a/pinot-plugins/pinot-input-format/pom.xml
b/pinot-plugins/pinot-input-format/pinot-bson/pom.xml
similarity index 60%
copy from pinot-plugins/pinot-input-format/pom.xml
copy to pinot-plugins/pinot-input-format/pinot-bson/pom.xml
index 43873ae79e8..b4dd0dd139d 100644
--- a/pinot-plugins/pinot-input-format/pom.xml
+++ b/pinot-plugins/pinot-input-format/pinot-bson/pom.xml
@@ -22,42 +22,32 @@
<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
<modelVersion>4.0.0</modelVersion>
<parent>
- <artifactId>pinot-plugins</artifactId>
+ <artifactId>pinot-input-format</artifactId>
<groupId>org.apache.pinot</groupId>
<version>1.6.0-SNAPSHOT</version>
</parent>
- <artifactId>pinot-input-format</artifactId>
- <packaging>pom</packaging>
- <name>Pinot Input Format</name>
+ <artifactId>pinot-bson</artifactId>
+ <name>Pinot BSON</name>
<url>https://pinot.apache.org/</url>
<properties>
- <pinot.root>${basedir}/../..</pinot.root>
- <plugin.type>pinot-input-format</plugin.type>
+ <pinot.root>${basedir}/../../..</pinot.root>
+ <shade.phase.prop>package</shade.phase.prop>
</properties>
- <modules>
- <module>pinot-arrow</module>
- <module>pinot-avro</module>
- <module>pinot-avro-base</module>
- <module>pinot-clp-log</module>
- <module>pinot-confluent-avro</module>
- <module>pinot-confluent-json</module>
- <module>pinot-confluent-protobuf</module>
- <module>pinot-orc</module>
- <module>pinot-json</module>
- <module>pinot-parquet</module>
- <module>pinot-csv</module>
- <module>pinot-thrift</module>
- <module>pinot-protobuf</module>
- </modules>
-
<dependencies>
<dependency>
- <groupId>org.apache.pinot</groupId>
- <artifactId>pinot-spi</artifactId>
- <type>test-jar</type>
- <scope>test</scope>
+ <groupId>org.mongodb</groupId>
+ <artifactId>bson</artifactId>
</dependency>
</dependencies>
+
+ <profiles>
+ <profile>
+ <id>pinot-fastdev</id>
+ <properties>
+ <shade.phase.prop>none</shade.phase.prop>
+ </properties>
+ </profile>
+ </profiles>
</project>
diff --git
a/pinot-plugins/pinot-input-format/pinot-bson/src/main/java/org/apache/pinot/plugin/inputformat/bson/BSONMessageDecoder.java
b/pinot-plugins/pinot-input-format/pinot-bson/src/main/java/org/apache/pinot/plugin/inputformat/bson/BSONMessageDecoder.java
new file mode 100644
index 00000000000..4a7df0b7c86
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-bson/src/main/java/org/apache/pinot/plugin/inputformat/bson/BSONMessageDecoder.java
@@ -0,0 +1,67 @@
+/**
+ * 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.pinot.plugin.inputformat.bson;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.data.readers.RecordExtractor;
+import org.apache.pinot.spi.plugin.PluginManager;
+import org.apache.pinot.spi.stream.StreamMessageDecoder;
+import org.bson.Document;
+
+
+/// An implementation of [StreamMessageDecoder] to read BSON records from a
stream. Each message payload is
+/// expected to be a single binary-encoded BSON document. The document is
decoded into a [Document] (a
+/// `Map<String, Object>`) and handed to a [BSONRecordExtractor].
+public class BSONMessageDecoder implements StreamMessageDecoder<byte[]> {
+ private static final String BSON_RECORD_EXTRACTOR_CLASS =
+ "org.apache.pinot.plugin.inputformat.bson.BSONRecordExtractor";
+
+ private RecordExtractor<Map<String, Object>> _bsonRecordExtractor;
+
+ @Override
+ public void init(Map<String, String> props, Set<String> fieldsToRead, String
topicName)
+ throws Exception {
+ String recordExtractorClass = null;
+ if (props != null) {
+ recordExtractorClass = props.get(RECORD_EXTRACTOR_CONFIG_KEY);
+ }
+ if (recordExtractorClass == null) {
+ recordExtractorClass = BSON_RECORD_EXTRACTOR_CLASS;
+ }
+ _bsonRecordExtractor =
PluginManager.get().createInstance(recordExtractorClass);
+ _bsonRecordExtractor.init(fieldsToRead, null);
+ }
+
+ @Override
+ public GenericRow decode(byte[] payload, GenericRow destination) {
+ return decode(payload, 0, payload.length, destination);
+ }
+
+ @Override
+ public GenericRow decode(byte[] payload, int offset, int length, GenericRow
destination) {
+ try {
+ Document document = BSONUtils.decodeDocument(payload, offset, length);
+ return _bsonRecordExtractor.extract(document, destination);
+ } catch (Exception e) {
+ throw new RuntimeException("Caught exception while decoding BSON
record", e);
+ }
+ }
+}
diff --git
a/pinot-plugins/pinot-input-format/pinot-bson/src/main/java/org/apache/pinot/plugin/inputformat/bson/BSONRecordExtractor.java
b/pinot-plugins/pinot-input-format/pinot-bson/src/main/java/org/apache/pinot/plugin/inputformat/bson/BSONRecordExtractor.java
new file mode 100644
index 00000000000..050e215520d
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-bson/src/main/java/org/apache/pinot/plugin/inputformat/bson/BSONRecordExtractor.java
@@ -0,0 +1,147 @@
+/**
+ * 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.pinot.plugin.inputformat.bson;
+
+import com.google.common.collect.Maps;
+import java.math.BigDecimal;
+import java.sql.Timestamp;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import org.apache.pinot.spi.data.readers.BaseRecordExtractor;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.bson.BsonTimestamp;
+import org.bson.types.Binary;
+import org.bson.types.Decimal128;
+import org.bson.types.ObjectId;
+
+
+/// Extracts a Pinot [GenericRow] from a decoded BSON document
(`org.bson.Document`, which is a
+/// `Map<String, Object>`). Values are the Java objects produced by the
standard MongoDB
+/// [org.bson.codecs.DocumentCodec].
+///
+/// **BSON type → Java output type:**
+/// - `Double` / `Int32` / `Int64` / `Boolean` / `String` → same boxed type
(pass-through)
+/// - `Document` (embedded) → `Map<String, Object>` (values recursively
converted)
+/// - `Array` → `Object[]` (elements recursively converted)
+/// - `ObjectId` → `String` (24-char hex)
+/// - `DateTime` → `java.sql.Timestamp`
+/// - `Decimal128` → `BigDecimal` (`NaN` / `Infinity` → `null`, as
`BigDecimal` cannot represent them; negative
+/// zero → `BigDecimal.ZERO`)
+/// - `Timestamp` (the internal replication type, e.g. the oplog /
change-stream `ts` and `clusterTime` fields)
+/// → `java.sql.Timestamp` at second granularity, matching MongoDB's own
`$toDate`. The seconds field is
+/// unsigned, so it is read as such and remains correct past 2038. The
ordinal counter that disambiguates
+/// operations within the same second is not representable in a `Timestamp`
and is dropped, so this value
+/// orders operations only to the second.
+/// - `Binary` → `byte[]`. This includes the UUID subtypes (`0x03` legacy,
`0x04` standard): the standard codec
+/// is constructed with `UuidRepresentation.UNSPECIFIED`, so it hands back
the raw 16 bytes rather than a
+/// `java.util.UUID`.
+/// - `null` → `null`
+///
+/// Every other BSON type falls back to `value.toString()` — the deprecated
`Symbol` / `Undefined` / `DBPointer`
+/// types, JavaScript `Code`, regular expressions, `MinKey` / `MaxKey`, and
the binary vector types
+/// (`Float32BinaryVector` and friends), which the standard codec decodes to
their own classes rather than to
+/// `Binary`. These renderings are not a documented contract of
`org.mongodb:bson`, so `BSONRecordExtractorTest`
+/// pins the exact strings: a driver upgrade that changes them must fail the
build rather than silently change
+/// the contents of every segment built from such a document.
+///
+/// The converted values follow the shared `RecordExtractor` contract, so the
downstream data-type transformer
+/// coerces them to the declared column type.
+public class BSONRecordExtractor extends BaseRecordExtractor<Map<String,
Object>> {
+
+ @Override
+ public GenericRow extract(Map<String, Object> from, GenericRow to) {
+ if (_extractAll) {
+ for (Map.Entry<String, Object> entry : from.entrySet()) {
+ Object value = entry.getValue();
+ to.putValue(entry.getKey(), value != null ? convert(value) : null);
+ }
+ } else {
+ for (String fieldName : _fields) {
+ Object value = from.get(fieldName);
+ to.putValue(fieldName, value != null ? convert(value) : null);
+ }
+ }
+ return to;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Object convert(Object value) {
+ if (value instanceof Map) {
+ return convertMap((Map<String, Object>) value);
+ }
+ if (value instanceof List) {
+ return convertList((List<Object>) value);
+ }
+ if (value instanceof ObjectId) {
+ return ((ObjectId) value).toHexString();
+ }
+ if (value instanceof Date) {
+ return new Timestamp(((Date) value).getTime());
+ }
+ if (value instanceof BsonTimestamp) {
+ // getTime() is the seconds component; the ordinal counter in the low 32
bits is dropped. BSON stores
+ // those seconds unsigned but getTime() returns a signed int, so mask
before widening -- otherwise every
+ // timestamp from 2038-01-19 onward wraps to a pre-1970 date.
+ return new Timestamp((((BsonTimestamp) value).getTime() & 0xFFFFFFFFL) *
1000L);
+ }
+ // NOTE: Decimal128 extends Number, so this must stay above the Number
pass-through below.
+ if (value instanceof Decimal128) {
+ Decimal128 decimal128 = (Decimal128) value;
+ // NaN / Infinity are legal Decimal128 values with no BigDecimal
representation; surface them as null.
+ if (decimal128.isNaN() || decimal128.isInfinite()) {
+ return null;
+ }
+ try {
+ return decimal128.bigDecimalValue();
+ } catch (ArithmeticException e) {
+ // The only remaining value bigDecimalValue() rejects is negative zero
(legal in BSON, at any exponent,
+ // and not detectable via a public predicate). It is numerically zero.
+ return BigDecimal.ZERO;
+ }
+ }
+ if (value instanceof Binary) {
+ return ((Binary) value).getData();
+ }
+ // Double / Integer / Long / String / Boolean pass through; any other BSON
type uses its string form.
+ if (value instanceof Number || value instanceof String || value instanceof
Boolean) {
+ return value;
+ }
+ return value.toString();
+ }
+
+ private static Object[] convertList(List<Object> list) {
+ int numValues = list.size();
+ Object[] result = new Object[numValues];
+ for (int i = 0; i < numValues; i++) {
+ Object value = list.get(i);
+ result[i] = value != null ? convert(value) : null;
+ }
+ return result;
+ }
+
+ private static Map<String, Object> convertMap(Map<String, Object> map) {
+ Map<String, Object> result = Maps.newHashMapWithExpectedSize(map.size());
+ for (Map.Entry<String, Object> entry : map.entrySet()) {
+ Object value = entry.getValue();
+ result.put(entry.getKey(), value != null ? convert(value) : null);
+ }
+ return result;
+ }
+}
diff --git
a/pinot-plugins/pinot-input-format/pinot-bson/src/main/java/org/apache/pinot/plugin/inputformat/bson/BSONRecordReader.java
b/pinot-plugins/pinot-input-format/pinot-bson/src/main/java/org/apache/pinot/plugin/inputformat/bson/BSONRecordReader.java
new file mode 100644
index 00000000000..d30854df337
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-bson/src/main/java/org/apache/pinot/plugin/inputformat/bson/BSONRecordReader.java
@@ -0,0 +1,168 @@
+/**
+ * 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.pinot.plugin.inputformat.bson;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.data.readers.RecordFetchException;
+import org.apache.pinot.spi.data.readers.RecordReader;
+import org.apache.pinot.spi.data.readers.RecordReaderConfig;
+import org.apache.pinot.spi.data.readers.RecordReaderUtils;
+import org.bson.Document;
+
+
+/// Record reader for a BSON file: a concatenation of framed BSON documents
(the `mongodump` layout), each
+/// self-delimited by a leading little-endian int32 byte length. Documents are
read sequentially, with the next
+/// document's bytes fetched ahead so [#hasNext] does not perform I/O.
GZIP-compressed files are supported.
+///
+/// Not thread-safe: a reader instance is single-threaded, like every other
[RecordReader].
+public class BSONRecordReader implements RecordReader {
+ // Minimum size of a BSON document: 4-byte length prefix + 1-byte
terminating NUL of an empty document.
+ private static final int MIN_DOCUMENT_LENGTH = 5;
+ // MongoDB caps BSON documents at 16MB. Without an upper bound, a corrupt
length prefix (up to ~2GB) would
+ // reach `new byte[length]` and raise OutOfMemoryError -- an Error, so it
would escape the recoverable
+ // exception handling in next()/advance() and abort the whole segment build.
+ private static final int MAX_DOCUMENT_LENGTH = 16 * 1024 * 1024;
+
+ private File _dataFile;
+ private BSONRecordExtractor _recordExtractor;
+ private InputStream _inputStream;
+ // Bytes of the next framed document, or null once the stream is exhausted.
+ private byte[] _nextDocument;
+ // A read error hit while fetching the next document. Deferred so the
current record is still emitted; it
+ // surfaces on the following next() call rather than discarding an
already-read valid record.
+ private IOException _fetchError;
+
+ public BSONRecordReader() {
+ }
+
+ @Override
+ public void init(File dataFile, @Nullable Set<String> fieldsToRead,
@Nullable RecordReaderConfig recordReaderConfig)
+ throws IOException {
+ _dataFile = dataFile;
+ _recordExtractor = new BSONRecordExtractor();
+ _recordExtractor.init(fieldsToRead, null);
+ open();
+ }
+
+ private void open()
+ throws IOException {
+ _inputStream = RecordReaderUtils.getBufferedInputStream(_dataFile);
+ _fetchError = null;
+ try {
+ _nextDocument = readNextDocument();
+ } catch (IOException e) {
+ _inputStream.close();
+ throw e;
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ return _nextDocument != null || _fetchError != null;
+ }
+
+ @Override
+ public GenericRow next(GenericRow reuse)
+ throws IOException {
+ if (_fetchError != null) {
+ IOException error = _fetchError;
+ _fetchError = null;
+ throw new RecordFetchException("Failed to read next BSON record", error);
+ }
+ byte[] documentBytes = _nextDocument;
+ Document document;
+ try {
+ document = BSONUtils.decodeDocument(documentBytes);
+ } catch (RuntimeException e) {
+ // Corrupt frame: advance past it (bytes already consumed) so we don't
retry, then report a parse error.
+ advance();
+ throw new IOException("Failed to decode BSON record", e);
+ }
+ _recordExtractor.extract(document, reuse);
+ advance();
+ return reuse;
+ }
+
+ /// Advances the look-ahead to the next framed document. A read error is
stashed rather than thrown so the
+ /// record just returned by next() is still emitted; the error surfaces on
the following next() call.
+ private void advance() {
+ try {
+ _nextDocument = readNextDocument();
+ } catch (IOException e) {
+ _nextDocument = null;
+ _fetchError = e;
+ }
+ }
+
+ /// Reads the next framed BSON document in full, or returns `null` at a
clean end-of-stream. Throws when the
+ /// stream ends partway through a document, or the length prefix is invalid.
+ @Nullable
+ private byte[] readNextDocument()
+ throws IOException {
+ int b0 = _inputStream.read();
+ if (b0 == -1) {
+ return null;
+ }
+ int b1 = _inputStream.read();
+ int b2 = _inputStream.read();
+ int b3 = _inputStream.read();
+ if ((b1 | b2 | b3) < 0) {
+ throw new IOException("Truncated BSON document: incomplete length
prefix");
+ }
+ // BSON length prefix is a little-endian int32 inclusive of these 4 bytes.
+ int length = (b0 & 0xFF) | ((b1 & 0xFF) << 8) | ((b2 & 0xFF) << 16) | ((b3
& 0xFF) << 24);
+ if (length < MIN_DOCUMENT_LENGTH || length > MAX_DOCUMENT_LENGTH) {
+ throw new IOException(
+ "Invalid BSON document length: " + length + " (expected " +
MIN_DOCUMENT_LENGTH + ".."
+ + MAX_DOCUMENT_LENGTH + " bytes)");
+ }
+ byte[] document = new byte[length];
+ document[0] = (byte) b0;
+ document[1] = (byte) b1;
+ document[2] = (byte) b2;
+ document[3] = (byte) b3;
+ int bytesRead = 4;
+ while (bytesRead < length) {
+ int count = _inputStream.read(document, bytesRead, length - bytesRead);
+ if (count == -1) {
+ throw new IOException("Truncated BSON document: expected " + length +
" bytes");
+ }
+ bytesRead += count;
+ }
+ return document;
+ }
+
+ @Override
+ public void rewind()
+ throws IOException {
+ _inputStream.close();
+ open();
+ }
+
+ @Override
+ public void close()
+ throws IOException {
+ _inputStream.close();
+ }
+}
diff --git
a/pinot-plugins/pinot-input-format/pinot-bson/src/main/java/org/apache/pinot/plugin/inputformat/bson/BSONUtils.java
b/pinot-plugins/pinot-input-format/pinot-bson/src/main/java/org/apache/pinot/plugin/inputformat/bson/BSONUtils.java
new file mode 100644
index 00000000000..3703ef2ca94
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-bson/src/main/java/org/apache/pinot/plugin/inputformat/bson/BSONUtils.java
@@ -0,0 +1,57 @@
+/**
+ * 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.pinot.plugin.inputformat.bson;
+
+import java.nio.ByteBuffer;
+import org.bson.BsonBinaryReader;
+import org.bson.Document;
+import org.bson.codecs.DecoderContext;
+import org.bson.codecs.DocumentCodec;
+
+
+/// Decodes a single BSON binary document into an `org.bson.Document` (a
`Map<String, Object>`). Shared by
+/// [BSONRecordReader] (batch) and [BSONMessageDecoder] (streaming).
+///
+/// Thread-safe: the shared [DocumentCodec] and [DecoderContext] are immutable
after construction, and each
+/// [#decodeDocument] call reads through a reader over its own buffer. Callers
on different ingestion threads
+/// may decode concurrently.
+///
+/// A malformed document raises `org.bson.BsonSerializationException` (a
`RuntimeException`). Bogus internal
+/// length prefixes cannot over-allocate: the reader is bounded by the
supplied `[offset, offset + length)`
+/// slice.
+public final class BSONUtils {
+ private static final DocumentCodec DOCUMENT_CODEC = new DocumentCodec();
+ private static final DecoderContext DECODER_CONTEXT =
DecoderContext.builder().build();
+
+ private BSONUtils() {
+ }
+
+ /// Decodes a whole BSON document from the given byte array.
+ public static Document decodeDocument(byte[] bytes) {
+ return decodeDocument(bytes, 0, bytes.length);
+ }
+
+ /// Decodes a single BSON document from the `[offset, offset + length)`
sub-range of the given byte array.
+ public static Document decodeDocument(byte[] bytes, int offset, int length) {
+ ByteBuffer buffer = ByteBuffer.wrap(bytes, offset, length).slice();
+ try (BsonBinaryReader reader = new BsonBinaryReader(buffer)) {
+ return DOCUMENT_CODEC.decode(reader, DECODER_CONTEXT);
+ }
+ }
+}
diff --git
a/pinot-plugins/pinot-input-format/pinot-bson/src/test/java/org/apache/pinot/plugin/inputformat/bson/BSONMessageDecoderTest.java
b/pinot-plugins/pinot-input-format/pinot-bson/src/test/java/org/apache/pinot/plugin/inputformat/bson/BSONMessageDecoderTest.java
new file mode 100644
index 00000000000..77a8ac2b031
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-bson/src/test/java/org/apache/pinot/plugin/inputformat/bson/BSONMessageDecoderTest.java
@@ -0,0 +1,153 @@
+/**
+ * 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.pinot.plugin.inputformat.bson;
+
+import java.math.BigDecimal;
+import java.sql.Timestamp;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.bson.BsonBinaryWriter;
+import org.bson.BsonTimestamp;
+import org.bson.Document;
+import org.bson.codecs.DocumentCodec;
+import org.bson.codecs.EncoderContext;
+import org.bson.io.BasicOutputBuffer;
+import org.bson.types.Decimal128;
+import org.bson.types.ObjectId;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
+
+
+public class BSONMessageDecoderTest {
+
+ @Test
+ public void testDecodeScalarsAndLogicalTypes()
+ throws Exception {
+ ObjectId objectId = new ObjectId("5f2b1c0e1c9d440000a1b2c3");
+ long epochMillis = 1_588_469_340_000L;
+ Document document = new Document()
+ .append("id", objectId)
+ .append("name", "widget")
+ .append("count", 7)
+ .append("views", 1_588_469_340_000L)
+ .append("price", 9.5d)
+ .append("active", true)
+ .append("amount", Decimal128.parse("123.456"))
+ .append("createdAt", new Date(epochMillis))
+ // The oplog / change-stream replication timestamp, the field a
Kafka-forwarded change stream is most
+ // likely to use as the Pinot time column.
+ .append("ts", new BsonTimestamp((int) (epochMillis / 1000), 3));
+
+ Set<String> fields = document.keySet();
+ BSONMessageDecoder decoder = new BSONMessageDecoder();
+ decoder.init(Map.of(), fields, "testTopic");
+
+ GenericRow row = new GenericRow();
+ decoder.decode(encode(document), row);
+
+ assertEquals(row.getValue("id"), objectId.toHexString());
+ assertEquals(row.getValue("name"), "widget");
+ assertEquals(row.getValue("count"), 7);
+ assertEquals(row.getValue("views"), 1_588_469_340_000L);
+ assertEquals(row.getValue("price"), 9.5d);
+ assertEquals(row.getValue("active"), true);
+ assertEquals(row.getValue("amount"), new BigDecimal("123.456"));
+ assertEquals(row.getValue("createdAt"), new Timestamp(epochMillis));
+ assertEquals(row.getValue("ts"), new Timestamp(epochMillis));
+ }
+
+ @Test
+ public void testDecodeNestedDocumentAndArray()
+ throws Exception {
+ Document document = new Document()
+ .append("tags", List.of("a", "b", "c"))
+ .append("meta", new Document("region", "us-east").append("shard", 3));
+
+ Set<String> fields = document.keySet();
+ BSONMessageDecoder decoder = new BSONMessageDecoder();
+ decoder.init(Map.of(), fields, "testTopic");
+
+ GenericRow row = new GenericRow();
+ decoder.decode(encode(document), row);
+
+ assertEquals((Object[]) row.getValue("tags"), new Object[]{"a", "b", "c"});
+ Map<?, ?> meta = (Map<?, ?>) row.getValue("meta");
+ assertEquals(meta.get("region"), "us-east");
+ assertEquals(meta.get("shard"), 3);
+ }
+
+ @Test
+ public void testDecodeWithFieldProjection()
+ throws Exception {
+ Document document = new Document().append("keep", 1).append("drop", 2);
+
+ BSONMessageDecoder decoder = new BSONMessageDecoder();
+ decoder.init(Map.of(), Set.of("keep"), "testTopic");
+
+ GenericRow row = new GenericRow();
+ decoder.decode(encode(document), row);
+
+ assertEquals(row.getValue("keep"), 1);
+ assertNull(row.getValue("drop"));
+ }
+
+ @Test
+ public void testDecodeWithOffsetAndLength()
+ throws Exception {
+ Document document = new Document().append("k", "v");
+ byte[] encoded = encode(document);
+
+ // Embed the payload inside a larger buffer to exercise the offset/length
decode path.
+ byte[] padded = new byte[encoded.length + 8];
+ int offset = 5;
+ System.arraycopy(encoded, 0, padded, offset, encoded.length);
+
+ BSONMessageDecoder decoder = new BSONMessageDecoder();
+ decoder.init(Map.of(), Set.of("k"), "testTopic");
+
+ GenericRow row = new GenericRow();
+ GenericRow result = decoder.decode(padded, offset, encoded.length, row);
+ assertSame(result, row);
+ assertEquals(row.getValue("k"), "v");
+ }
+
+ @Test(expectedExceptions = RuntimeException.class)
+ public void testDecodeCorruptPayloadThrows()
+ throws Exception {
+ BSONMessageDecoder decoder = new BSONMessageDecoder();
+ decoder.init(Map.of(), Set.of("k"), "testTopic");
+ // Not a valid framed BSON document: the decode failure is wrapped in a
RuntimeException.
+ decoder.decode(new byte[]{1, 2, 3}, new GenericRow());
+ }
+
+ private static byte[] encode(Document document) {
+ DocumentCodec codec = new DocumentCodec();
+ BasicOutputBuffer buffer = new BasicOutputBuffer();
+ try (BsonBinaryWriter writer = new BsonBinaryWriter(buffer)) {
+ codec.encode(writer, document, EncoderContext.builder().build());
+ }
+ return buffer.toByteArray();
+ }
+}
diff --git
a/pinot-plugins/pinot-input-format/pinot-bson/src/test/java/org/apache/pinot/plugin/inputformat/bson/BSONRecordExtractorTest.java
b/pinot-plugins/pinot-input-format/pinot-bson/src/test/java/org/apache/pinot/plugin/inputformat/bson/BSONRecordExtractorTest.java
new file mode 100644
index 00000000000..ae2ac2ce9e0
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-bson/src/test/java/org/apache/pinot/plugin/inputformat/bson/BSONRecordExtractorTest.java
@@ -0,0 +1,290 @@
+/**
+ * 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.pinot.plugin.inputformat.bson;
+
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.sql.Timestamp;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.regex.Pattern;
+import javax.annotation.Nullable;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.bson.BsonBinaryWriter;
+import org.bson.BsonTimestamp;
+import org.bson.Document;
+import org.bson.codecs.DocumentCodec;
+import org.bson.codecs.EncoderContext;
+import org.bson.io.BasicOutputBuffer;
+import org.bson.types.Binary;
+import org.bson.types.Code;
+import org.bson.types.Decimal128;
+import org.bson.types.MaxKey;
+import org.bson.types.MinKey;
+import org.bson.types.ObjectId;
+import org.bson.types.Symbol;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Tests [BSONRecordExtractor] — see its class Javadoc for the BSON type →
Java output type matrix.
+public class BSONRecordExtractorTest {
+
+ private static final String COLUMN = "col";
+
+ // === Scalars — pass through ===
+
+ @Test
+ public void testBooleanPreserved() {
+ assertEquals(extract(true), true);
+ assertEquals(extract(false), false);
+ }
+
+ @Test
+ public void testIntegerPreserved() {
+ assertEquals(extract(42), 42);
+ }
+
+ @Test
+ public void testLongPreserved() {
+ assertEquals(extract(1_588_469_340_000L), 1_588_469_340_000L);
+ }
+
+ @Test
+ public void testDoublePreserved() {
+ assertEquals(extract(1.5d), 1.5d);
+ }
+
+ @Test
+ public void testStringPreserved() {
+ assertEquals(extract("hello"), "hello");
+ }
+
+ @Test
+ public void testNullPassedThrough() {
+ assertNull(extract(null));
+ }
+
+ // === Extended BSON types found in normal documents ===
+
+ @Test
+ public void testObjectIdConvertedToHexString() {
+ ObjectId objectId = new ObjectId("5f2b1c0e1c9d440000a1b2c3");
+ assertEquals(extract(objectId), "5f2b1c0e1c9d440000a1b2c3");
+ }
+
+ @Test
+ public void testDateConvertedToTimestamp() {
+ long epochMillis = 1_588_469_340_000L;
+ Object result = extract(new Date(epochMillis));
+ assertTrue(result instanceof Timestamp);
+ assertEquals(result, new Timestamp(epochMillis));
+ }
+
+ @Test
+ public void testDecimal128ConvertedToBigDecimal() {
+ assertEquals(extract(Decimal128.parse("123.456")), new
BigDecimal("123.456"));
+ }
+
+ @Test
+ public void testDecimal128NaNConvertedToNull() {
+ // NaN / Infinity are legal Decimal128 values with no BigDecimal
representation, so they surface as null.
+ assertNull(extract(Decimal128.parse("NaN")));
+ assertNull(extract(Decimal128.parse("Infinity")));
+ assertNull(extract(Decimal128.parse("-Infinity")));
+ }
+
+ @Test
+ public void testDecimal128NegativeZeroConvertedToZero() {
+ // Negative zero is a legal Decimal128 value that bigDecimalValue()
rejects; it is numerically zero. It is
+ // negative-zero at any exponent, so "-0.00" must be handled too (it is
not equal to Decimal128.NEGATIVE_ZERO).
+ assertEquals(extract(Decimal128.parse("-0")), BigDecimal.ZERO);
+ assertEquals(extract(Decimal128.parse("-0.00")), BigDecimal.ZERO);
+ }
+
+ @Test
+ public void testBinaryConvertedToByteArray() {
+ byte[] data = {1, 2, 3, 4};
+ Object result = extract(new Binary(data));
+ assertTrue(result instanceof byte[]);
+ assertEquals((byte[]) result, data);
+ }
+
+ @Test
+ public void testBsonTimestampConvertedToTimestamp() {
+ // The internal replication timestamp: the oplog `ts` field and the
change-stream `clusterTime` field. Its
+ // seconds component becomes a java.sql.Timestamp; the intra-second
ordinal counter is dropped.
+ Object result = extract(new BsonTimestamp(1_588_469_340, 7));
+ assertTrue(result instanceof Timestamp);
+ assertEquals(result, new Timestamp(1_588_469_340_000L));
+ }
+
+ @Test
+ public void testBsonTimestampAfter2038IsNotSignWrapped() {
+ // BSON stores the seconds component as an UNSIGNED 32-bit field, but
BsonTimestamp#getTime() returns it as
+ // a signed int. From 2038-01-19T03:14:08Z onward the high bit is set, so
a naive read wraps to a pre-1970
+ // date. 2039-01-01T00:00:00Z = 2_177_452_800 seconds, which does not fit
a positive signed int.
+ long seconds = 2_177_452_800L;
+ BsonTimestamp afterY2038 = new BsonTimestamp((int) seconds, 1);
+ assertTrue(afterY2038.getTime() < 0, "precondition: the driver reports
these seconds as a negative int");
+
+ Object result = extract(afterY2038);
+ assertEquals(result, new Timestamp(seconds * 1000L));
+ }
+
+ @Test
+ public void testUuidBinarySubtypesConvertedToRawBytes() {
+ // Subtypes 0x03 (legacy) and 0x04 (standard) are how MongoDB stores
UUIDs. The standard DocumentCodec is
+ // built with UuidRepresentation.UNSPECIFIED, so it decodes them to Binary
-- not java.util.UUID -- and they
+ // land as raw 16-byte BYTES. Round-trip through a real encode/decode so
this pins the codec's behavior
+ // rather than our own hand-built Binary: if a driver upgrade flips the
default representation, these
+ // columns would silently change from BYTES to a 36-char STRING.
+ UUID uuid = UUID.fromString("123e4567-e89b-12d3-a456-426614174000");
+ byte[] raw = new byte[16];
+
ByteBuffer.wrap(raw).putLong(uuid.getMostSignificantBits()).putLong(uuid.getLeastSignificantBits());
+
+ Document decoded = roundTrip(new Document()
+ .append("standard", new Binary((byte) 0x04, raw))
+ .append("legacy", new Binary((byte) 0x03, raw)));
+
+ GenericRow row = new GenericRow();
+ newExtractor().extract(decoded, row);
+ assertEquals((byte[]) row.getValue("standard"), raw);
+ assertEquals((byte[]) row.getValue("legacy"), raw);
+ }
+
+ @Test
+ public void testExoticTypesFallBackToPinnedToStringForms() {
+ // Rare / deprecated types with no Pinot-native representation fall back
to value.toString(). Those strings
+ // are not a documented contract of org.mongodb:bson, and they get baked
into segment data -- so pin them
+ // literally. A driver upgrade that reformats them must fail here rather
than silently rewrite segments.
+ assertEquals(extract(new MaxKey()), "MaxKey");
+ assertEquals(extract(new MinKey()), "MinKey");
+ assertEquals(extract(new Symbol("sym")), "sym");
+ assertEquals(extract(new Code("function(){}")),
"Code{code='function(){}'}");
+
+ // Regex and JS code survive an encode/decode round trip as
BsonRegularExpression / Code.
+ Document decoded = roundTrip(new Document()
+ .append("regex", Pattern.compile("^a.*z$"))
+ .append("code", new Code("function(){}")));
+ GenericRow row = new GenericRow();
+ newExtractor().extract(decoded, row);
+ assertEquals(row.getValue("regex"),
"BsonRegularExpression{pattern='^a.*z$', options=''}");
+ assertEquals(row.getValue("code"), "Code{code='function(){}'}");
+ }
+
+ // === Array (BSON array) → Object[] ===
+
+ @Test
+ public void testIntegerListExtractedAsArray() {
+ Object[] result = (Object[]) extract(List.of(10, 20, 30));
+ assertEquals(result, new Object[]{10, 20, 30});
+ }
+
+ @Test
+ public void testStringListExtractedAsArray() {
+ Object[] result = (Object[]) extract(List.of("foo", "bar"));
+ assertEquals(result, new Object[]{"foo", "bar"});
+ }
+
+ @Test
+ public void testListWithNullElement() {
+ Object[] result = (Object[]) extract(Arrays.asList(1, null, 3));
+ assertEquals(result, new Object[]{1, null, 3});
+ }
+
+ @Test
+ public void testEmptyListExtractedAsEmptyArray() {
+ Object[] result = (Object[]) extract(List.of());
+ assertEquals(result, new Object[]{});
+ }
+
+ @Test
+ public void testListOfObjectIdsConvertedElementwise() {
+ ObjectId id1 = new ObjectId("5f2b1c0e1c9d440000a1b2c3");
+ ObjectId id2 = new ObjectId("5f2b1c0e1c9d440000a1b2c4");
+ Object[] result = (Object[]) extract(List.of(id1, id2));
+ assertEquals(result, new Object[]{id1.toHexString(), id2.toHexString()});
+ }
+
+ // === Embedded document (BSON object) → Map<String, Object> ===
+
+ @Test
+ public void testEmbeddedDocumentConvertedToMap() {
+ Document embedded = new Document("k1", 1).append("k2", "foo").append("k3",
true);
+ Map<?, ?> result = (Map<?, ?>) extract(embedded);
+ assertEquals(result.size(), 3);
+ assertEquals(result.get("k1"), 1);
+ assertEquals(result.get("k2"), "foo");
+ assertEquals(result.get("k3"), true);
+ }
+
+ @Test
+ public void testEmbeddedDocumentWithNullValue() {
+ Document embedded = new Document("present", 1);
+ embedded.put("absent", null);
+ Map<?, ?> result = (Map<?, ?>) extract(embedded);
+ assertEquals(result.get("present"), 1);
+ assertTrue(result.containsKey("absent"));
+ assertNull(result.get("absent"));
+ }
+
+ @Test
+ public void testNestedDocumentAndArray() {
+ Document embedded =
+ new Document("scalar", 1).append("nested", new Document("sub",
1.1)).append("list", List.of("a", "b"));
+ Map<?, ?> result = (Map<?, ?>) extract(embedded);
+ assertEquals(result.get("scalar"), 1);
+ assertEquals(((Map<?, ?>) result.get("nested")).get("sub"), 1.1);
+ assertEquals((Object[]) result.get("list"), new Object[]{"a", "b"});
+ }
+
+ // === Helpers ===
+
+ /// Run the extractor against a single-column input document and return the
extracted value.
+ private static Object extract(@Nullable Object input) {
+ Document record = new Document();
+ record.put(COLUMN, input);
+ GenericRow row = new GenericRow();
+ newExtractor().extract(record, row);
+ return row.getValue(COLUMN);
+ }
+
+ private static BSONRecordExtractor newExtractor() {
+ BSONRecordExtractor extractor = new BSONRecordExtractor();
+ extractor.init(null, null);
+ return extractor;
+ }
+
+ /// Encodes and re-decodes a document so the test sees the Java types the
standard `DocumentCodec` actually
+ /// produces, rather than the ones the test handed in.
+ private static Document roundTrip(Document document) {
+ BasicOutputBuffer buffer = new BasicOutputBuffer();
+ try (BsonBinaryWriter writer = new BsonBinaryWriter(buffer)) {
+ new DocumentCodec().encode(writer, document,
EncoderContext.builder().build());
+ }
+ return BSONUtils.decodeDocument(buffer.toByteArray());
+ }
+}
diff --git
a/pinot-plugins/pinot-input-format/pinot-bson/src/test/java/org/apache/pinot/plugin/inputformat/bson/BSONRecordReaderTest.java
b/pinot-plugins/pinot-input-format/pinot-bson/src/test/java/org/apache/pinot/plugin/inputformat/bson/BSONRecordReaderTest.java
new file mode 100644
index 00000000000..c3b8bbc9d82
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-bson/src/test/java/org/apache/pinot/plugin/inputformat/bson/BSONRecordReaderTest.java
@@ -0,0 +1,212 @@
+/**
+ * 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.pinot.plugin.inputformat.bson;
+
+import java.io.BufferedOutputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.spi.data.readers.AbstractRecordReaderTest;
+import org.apache.pinot.spi.data.readers.RecordReader;
+import org.bson.BsonBinaryWriter;
+import org.bson.Document;
+import org.bson.codecs.DocumentCodec;
+import org.bson.codecs.EncoderContext;
+import org.bson.io.BasicOutputBuffer;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+/// Tests [BSONRecordReader] against the shared [AbstractRecordReaderTest]
contract (happy path + gzip + rewind
+/// over the generated schema), plus targeted negative / recovery cases driven
by hand-built raw byte frames.
+///
+/// The generated `FLOAT` values are written as BSON doubles (BSON has no
32-bit float type), so the reader
+/// reads them back as `Double`; the base assertions compare floating values
with an epsilon, so this is
+/// transparent.
+public class BSONRecordReaderTest extends AbstractRecordReaderTest {
+
+ @Override
+ protected RecordReader createRecordReader(File file)
+ throws Exception {
+ BSONRecordReader recordReader = new BSONRecordReader();
+ recordReader.init(file, _sourceFields, null);
+ return recordReader;
+ }
+
+ @Override
+ protected void writeRecordsToFile(List<Map<String, Object>> recordsToWrite)
+ throws Exception {
+ try (OutputStream out = new BufferedOutputStream(new
FileOutputStream(_dataFile))) {
+ for (Map<String, Object> record : recordsToWrite) {
+ Document document = new Document();
+ for (Map.Entry<String, Object> entry : record.entrySet()) {
+ document.put(entry.getKey(), toBsonValue(entry.getValue()));
+ }
+ out.write(encode(document));
+ }
+ }
+ }
+
+ /// BSON's codec has no float codec, so normalize Float (and Floats inside
arrays) to Double before encoding.
+ private static Object toBsonValue(Object value) {
+ if (value instanceof Float) {
+ return ((Float) value).doubleValue();
+ }
+ if (value instanceof List) {
+ List<Object> converted = new ArrayList<>();
+ for (Object element : (List<?>) value) {
+ converted.add(toBsonValue(element));
+ }
+ return converted;
+ }
+ return value;
+ }
+
+ @Override
+ protected String getDataFileName() {
+ return "data.bson";
+ }
+
+ // === Negative / recovery cases over hand-built raw frames ===
+
+ @Test
+ public void testEmptyFileHasNoRecords()
+ throws Exception {
+ try (BSONRecordReader reader = readerOver("empty.bson", new byte[0])) {
+ assertFalse(reader.hasNext());
+ }
+ }
+
+ @Test
+ public void testTruncatedLengthPrefixThrows()
+ throws Exception {
+ // Only 2 of the 4 length-prefix bytes are present.
+ assertInitThrowsIOException("truncated-prefix.bson", new byte[]{0x0C,
0x00});
+ }
+
+ @Test
+ public void testLengthBelowMinimumThrows()
+ throws Exception {
+ // Declared length 2 is below the 5-byte minimum for a BSON document.
+ assertInitThrowsIOException("invalid-length.bson", new byte[]{0x02, 0x00,
0x00, 0x00});
+ }
+
+ @Test
+ public void testLengthAboveMaximumThrowsInsteadOfAllocating()
+ throws Exception {
+ // Declared length 0x7FFFFFFF (~2GB) exceeds the 16MB BSON maximum. It
must be rejected before `new
+ // byte[length]` runs, otherwise an OutOfMemoryError (an Error) would
escape the recoverable error paths.
+ assertInitThrowsIOException("oversized-length.bson", new byte[]{(byte)
0xFF, (byte) 0xFF, (byte) 0xFF, 0x7F});
+ }
+
+ @Test
+ public void testTruncatedBodyThrows()
+ throws Exception {
+ // Declared length 16 but only 2 body bytes follow the prefix.
+ assertInitThrowsIOException("truncated-body.bson", new byte[]{0x10, 0x00,
0x00, 0x00, 0x01, 0x02});
+ }
+
+ @Test
+ public void testRecoversAfterCorruptRecord()
+ throws Exception {
+ // A corrupt frame between two valid documents: length 8, then an unknown
BSON element type (0x1F).
+ byte[] corruptFrame = {0x08, 0x00, 0x00, 0x00, 0x1F, 0x00, 0x00, 0x00};
+ byte[] content =
+ concat(encode(new Document("a", 1)), corruptFrame, encode(new
Document("b", 2)));
+ try (BSONRecordReader reader = readerOver("recovery.bson", content)) {
+ assertTrue(reader.hasNext());
+ assertEquals(reader.next().getValue("a"), 1);
+
+ // The corrupt frame surfaces as a skippable IOException; the reader
then advances past it.
+ assertTrue(reader.hasNext());
+ assertThrows(IOException.class, reader::next);
+
+ // The trailing valid document is still returned.
+ assertTrue(reader.hasNext());
+ assertEquals(reader.next().getValue("b"), 2);
+ assertFalse(reader.hasNext());
+ }
+ }
+
+ @Test
+ public void testValidRecordEmittedBeforeTruncatedTail()
+ throws Exception {
+ // A complete document followed by a truncated frame (declares 16 bytes,
only 2 body bytes present). The
+ // valid record is still emitted; the truncation surfaces on the following
next() call.
+ byte[] truncatedFrame = {0x10, 0x00, 0x00, 0x00, 0x01, 0x02};
+ byte[] content = concat(encode(new Document("a", 1)), truncatedFrame);
+ try (BSONRecordReader reader = readerOver("truncated-tail.bson", content))
{
+ assertTrue(reader.hasNext());
+ assertEquals(reader.next().getValue("a"), 1);
+
+ assertTrue(reader.hasNext());
+ assertThrows(IOException.class, reader::next);
+ assertFalse(reader.hasNext());
+ }
+ }
+
+ private BSONRecordReader readerOver(String name, byte[] content)
+ throws IOException {
+ BSONRecordReader reader = new BSONRecordReader();
+ reader.init(writeTempFile(name, content), null, null);
+ return reader;
+ }
+
+ private void assertInitThrowsIOException(String name, byte[] content)
+ throws IOException {
+ File file = writeTempFile(name, content);
+ BSONRecordReader reader = new BSONRecordReader();
+ assertThrows(IOException.class, () -> reader.init(file, null, null));
+ }
+
+ private File writeTempFile(String name, byte[] content)
+ throws IOException {
+ File file = new File(_tempDir, name);
+ FileUtils.writeByteArrayToFile(file, content);
+ return file;
+ }
+
+ private static byte[] encode(Document document) {
+ DocumentCodec codec = new DocumentCodec();
+ BasicOutputBuffer buffer = new BasicOutputBuffer();
+ try (BsonBinaryWriter writer = new BsonBinaryWriter(buffer)) {
+ codec.encode(writer, document, EncoderContext.builder().build());
+ }
+ return buffer.toByteArray();
+ }
+
+ private static byte[] concat(byte[]... chunks)
+ throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ for (byte[] chunk : chunks) {
+ out.write(chunk);
+ }
+ return out.toByteArray();
+ }
+}
diff --git
a/pinot-plugins/pinot-input-format/pinot-bson/src/test/resources/log4j2.xml
b/pinot-plugins/pinot-input-format/pinot-bson/src/test/resources/log4j2.xml
new file mode 100644
index 00000000000..439331f9d7b
--- /dev/null
+++ b/pinot-plugins/pinot-input-format/pinot-bson/src/test/resources/log4j2.xml
@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<Configuration>
+ <Appenders>
+ <Console name="console" target="SYSTEM_OUT">
+ <PatternLayout pattern="%d{HH:mm:ss.SSS} %p [%c{1}] [%t] %m%n"/>
+ </Console>
+ </Appenders>
+ <Loggers>
+ <Logger name="org.apache.pinot" level="warn" additivity="false">
+ <AppenderRef ref="console"/>
+ </Logger>
+ <Root level="error">
+ <AppenderRef ref="console"/>
+ </Root>
+ </Loggers>
+</Configuration>
diff --git a/pinot-plugins/pinot-input-format/pom.xml
b/pinot-plugins/pinot-input-format/pom.xml
index 43873ae79e8..57b453db90d 100644
--- a/pinot-plugins/pinot-input-format/pom.xml
+++ b/pinot-plugins/pinot-input-format/pom.xml
@@ -40,6 +40,7 @@
<module>pinot-arrow</module>
<module>pinot-avro</module>
<module>pinot-avro-base</module>
+ <module>pinot-bson</module>
<module>pinot-clp-log</module>
<module>pinot-confluent-avro</module>
<module>pinot-confluent-json</module>
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/FileFormat.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/FileFormat.java
index 588f7fe2b39..722e5d7a6af 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/FileFormat.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/FileFormat.java
@@ -19,7 +19,7 @@
package org.apache.pinot.spi.data.readers;
public enum FileFormat {
- AVRO, GZIPPED_AVRO, CSV, JSON, PINOT, THRIFT, PARQUET, ORC, PROTO, ARROW,
OTHER;
+ AVRO, GZIPPED_AVRO, CSV, JSON, PINOT, THRIFT, PARQUET, ORC, PROTO, ARROW,
BSON, OTHER;
/**
* Converts an input format string to the corresponding FileFormat enum.
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/RecordReaderFactory.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/RecordReaderFactory.java
index 0500a9f4f15..a78730c44ae 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/RecordReaderFactory.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/RecordReaderFactory.java
@@ -44,6 +44,7 @@ public class RecordReaderFactory {
static final String DEFAULT_CSV_RECORD_READER_CONFIG_CLASS =
"org.apache.pinot.plugin.inputformat.csv.CSVRecordReaderConfig";
static final String DEFAULT_JSON_RECORD_READER_CLASS =
"org.apache.pinot.plugin.inputformat.json.JSONRecordReader";
+ static final String DEFAULT_BSON_RECORD_READER_CLASS =
"org.apache.pinot.plugin.inputformat.bson.BSONRecordReader";
static final String DEFAULT_THRIFT_RECORD_READER_CLASS =
"org.apache.pinot.plugin.inputformat.thrift.ThriftRecordReader";
static final String DEFAULT_THRIFT_RECORD_READER_CONFIG_CLASS =
@@ -76,6 +77,7 @@ public class RecordReaderFactory {
register(FileFormat.GZIPPED_AVRO, DEFAULT_AVRO_RECORD_READER_CLASS,
DEFAULT_AVRO_RECORD_READER_CONFIG_CLASS);
register(FileFormat.CSV, DEFAULT_CSV_RECORD_READER_CLASS,
DEFAULT_CSV_RECORD_READER_CONFIG_CLASS);
register(FileFormat.JSON, DEFAULT_JSON_RECORD_READER_CLASS, null);
+ register(FileFormat.BSON, DEFAULT_BSON_RECORD_READER_CLASS, null);
register(FileFormat.THRIFT, DEFAULT_THRIFT_RECORD_READER_CLASS,
DEFAULT_THRIFT_RECORD_READER_CONFIG_CLASS);
register(FileFormat.ORC, DEFAULT_ORC_RECORD_READER_CLASS, null);
register(FileFormat.PARQUET, DEFAULT_PARQUET_RECORD_READER_CLASS,
DEFAULT_PARQUET_RECORD_READER_CONFIG_CLASS);
diff --git
a/pinot-spi/src/test/java/org/apache/pinot/spi/data/readers/RecordReaderFactoryTest.java
b/pinot-spi/src/test/java/org/apache/pinot/spi/data/readers/RecordReaderFactoryTest.java
index 7c423727faf..657d7e9d115 100644
---
a/pinot-spi/src/test/java/org/apache/pinot/spi/data/readers/RecordReaderFactoryTest.java
+++
b/pinot-spi/src/test/java/org/apache/pinot/spi/data/readers/RecordReaderFactoryTest.java
@@ -34,6 +34,7 @@ public class RecordReaderFactoryTest {
assertEquals(getRecordReaderClassName("gzipped_avro"),
DEFAULT_AVRO_RECORD_READER_CLASS);
assertEquals(getRecordReaderClassName("csv"),
DEFAULT_CSV_RECORD_READER_CLASS);
assertEquals(getRecordReaderClassName("json"),
DEFAULT_JSON_RECORD_READER_CLASS);
+ assertEquals(getRecordReaderClassName("bson"),
DEFAULT_BSON_RECORD_READER_CLASS);
assertEquals(getRecordReaderClassName("thrift"),
DEFAULT_THRIFT_RECORD_READER_CLASS);
assertEquals(getRecordReaderClassName("orc"),
DEFAULT_ORC_RECORD_READER_CLASS);
assertEquals(getRecordReaderClassName("parquet"),
DEFAULT_PARQUET_RECORD_READER_CLASS);
@@ -46,6 +47,7 @@ public class RecordReaderFactoryTest {
assertEquals(getRecordReaderConfigClassName("gzipped_avro"),
DEFAULT_AVRO_RECORD_READER_CONFIG_CLASS);
assertEquals(getRecordReaderConfigClassName("csv"),
DEFAULT_CSV_RECORD_READER_CONFIG_CLASS);
assertNull(getRecordReaderConfigClassName("json"));
+ assertNull(getRecordReaderConfigClassName("bson"));
assertEquals(getRecordReaderConfigClassName("thrift"),
DEFAULT_THRIFT_RECORD_READER_CONFIG_CLASS);
assertNull(getRecordReaderConfigClassName("orc"));
assertEquals(getRecordReaderConfigClassName("parquet"),
DEFAULT_PARQUET_RECORD_READER_CONFIG_CLASS);
@@ -115,6 +117,7 @@ public class RecordReaderFactoryTest {
assertNotNull(DEFAULT_AVRO_RECORD_READER_CLASS);
assertNotNull(DEFAULT_CSV_RECORD_READER_CLASS);
assertNotNull(DEFAULT_JSON_RECORD_READER_CLASS);
+ assertNotNull(DEFAULT_BSON_RECORD_READER_CLASS);
assertNotNull(DEFAULT_THRIFT_RECORD_READER_CLASS);
assertNotNull(DEFAULT_ORC_RECORD_READER_CLASS);
assertNotNull(DEFAULT_PARQUET_RECORD_READER_CLASS);
@@ -124,6 +127,7 @@ public class RecordReaderFactoryTest {
assertEquals(DEFAULT_AVRO_RECORD_READER_CLASS,
"org.apache.pinot.plugin.inputformat.avro.AvroRecordReader");
assertEquals(DEFAULT_CSV_RECORD_READER_CLASS,
"org.apache.pinot.plugin.inputformat.csv.CSVRecordReader");
assertEquals(DEFAULT_JSON_RECORD_READER_CLASS,
"org.apache.pinot.plugin.inputformat.json.JSONRecordReader");
+ assertEquals(DEFAULT_BSON_RECORD_READER_CLASS,
"org.apache.pinot.plugin.inputformat.bson.BSONRecordReader");
assertEquals(DEFAULT_PROTO_RECORD_READER_CLASS,
"org.apache.pinot.plugin.inputformat.protobuf.ProtoBufRecordReader");
}
diff --git a/pinot-tools/pom.xml b/pinot-tools/pom.xml
index 13d5887d0ff..7364716bc81 100644
--- a/pinot-tools/pom.xml
+++ b/pinot-tools/pom.xml
@@ -77,6 +77,10 @@
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-json</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-bson</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-orc</artifactId>
diff --git a/pom.xml b/pom.xml
index d52001301fe..d1a540996fc 100644
--- a/pom.xml
+++ b/pom.xml
@@ -156,6 +156,7 @@
<arrow.version>19.0.0</arrow.version>
<avro.version>1.12.1</avro.version>
+ <bson.version>5.6.1</bson.version>
<parquet.version>1.17.1</parquet.version>
<orc.version>1.9.9</orc.version>
<hive.version>2.8.1</hive.version>
@@ -777,6 +778,11 @@
<artifactId>avro</artifactId>
<version>${avro.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.mongodb</groupId>
+ <artifactId>bson</artifactId>
+ <version>${bson.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro-mapred</artifactId>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
