feat(json): expose NdJsonReadOptions via registerJson and readJson

core/src/main/java/org/apache/datafusion/CsvReadOptions.java

@@ -33,15 +33,6 @@
  */
 public final class CsvReadOptions {
 
-  /** Compression of the file. Names match DataFusion's {@code FileCompressionType} variants. */
-  public enum FileCompressionType {
-    UNCOMPRESSED,
-    GZIP,
-    BZIP2,
-    XZ,
-    ZSTD
-  }
-
   private boolean hasHeader = true;
   private byte delimiter = (byte) ',';
   private byte quote = (byte) '"';
@@ -90,6 +81,9 @@ public CsvReadOptions newlinesInValues(boolean v) {
   }
 
   public CsvReadOptions schemaInferMaxRecords(long n) {
+    if (n < 0) {
+      throw new IllegalArgumentException("schemaInferMaxRecords must be non-negative, got " + n);
+    }
     this.schemaInferMaxRecords = n;
     return this;
   }
@@ -116,7 +110,7 @@ byte[] toBytes() {
             .setDelimiter(delimiter & 0xFF)
             .setQuote(quote & 0xFF)
             .setFileExtension(fileExtension)
-            .setFileCompressionType(toProto(fileCompressionType));
+            .setFileCompressionType(FileCompressionTypes.toProto(fileCompressionType));
     if (terminator != null) {
       b.setTerminator(terminator & 0xFF);
     }
@@ -138,22 +132,4 @@ byte[] toBytes() {
   Schema schema() {
     return schema;
   }
-
-  private static org.apache.datafusion.protobuf.FileCompressionType toProto(FileCompressionType t) {
-    switch (t) {
-      case UNCOMPRESSED:
-        return org.apache.datafusion.protobuf.FileCompressionType
-            .FILE_COMPRESSION_TYPE_UNCOMPRESSED;
-      case GZIP:
-        return org.apache.datafusion.protobuf.FileCompressionType.FILE_COMPRESSION_TYPE_GZIP;
-      case BZIP2:
-        return org.apache.datafusion.protobuf.FileCompressionType.FILE_COMPRESSION_TYPE_BZIP2;
-      case XZ:
-        return org.apache.datafusion.protobuf.FileCompressionType.FILE_COMPRESSION_TYPE_XZ;
-      case ZSTD:
-        return org.apache.datafusion.protobuf.FileCompressionType.FILE_COMPRESSION_TYPE_ZSTD;
-      default:
-        throw new IllegalArgumentException("unhandled FileCompressionType: " + t);
-    }
-  }
 }

core/src/main/java/org/apache/datafusion/FileCompressionType.java

@@ -0,0 +1,33 @@
+/*
+ * 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.datafusion;
+
+/**
+ * Compression of a file-format source. Shared by {@link CsvReadOptions} and {@link
+ * NdJsonReadOptions} (and any future format that exposes the same set of compressions). Variant
+ * names match DataFusion's Rust {@code FileCompressionType} so they round-trip across JNI.
+ */
+public enum FileCompressionType {
+  UNCOMPRESSED,
+  GZIP,
+  BZIP2,
+  XZ,
+  ZSTD
+}

core/src/main/java/org/apache/datafusion/FileCompressionTypes.java

@@ -0,0 +1,47 @@
+/*
+ * 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.datafusion;
+
+/**
+ * Internal helpers for translating between the user-facing {@link FileCompressionType} enum and the
+ * protobuf-generated {@code org.apache.datafusion.protobuf.FileCompressionType} enum used on the
+ * wire. Same variant set, different Java types.
+ */
+final class FileCompressionTypes {
+  private FileCompressionTypes() {}
+
+  static org.apache.datafusion.protobuf.FileCompressionType toProto(FileCompressionType t) {
+    switch (t) {
+      case UNCOMPRESSED:
+        return org.apache.datafusion.protobuf.FileCompressionType
+            .FILE_COMPRESSION_TYPE_UNCOMPRESSED;
+      case GZIP:
+        return org.apache.datafusion.protobuf.FileCompressionType.FILE_COMPRESSION_TYPE_GZIP;
+      case BZIP2:
+        return org.apache.datafusion.protobuf.FileCompressionType.FILE_COMPRESSION_TYPE_BZIP2;
+      case XZ:
+        return org.apache.datafusion.protobuf.FileCompressionType.FILE_COMPRESSION_TYPE_XZ;
+      case ZSTD:
+        return org.apache.datafusion.protobuf.FileCompressionType.FILE_COMPRESSION_TYPE_ZSTD;
+      default:
+        throw new IllegalArgumentException("unhandled FileCompressionType: " + t);
+    }
+  }
+}

core/src/main/java/org/apache/datafusion/NdJsonReadOptions.java

@@ -0,0 +1,78 @@
+/*
+ * 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.datafusion;
+
+import org.apache.arrow.vector.types.pojo.Schema;
+
+/**
+ * Configuration knobs for newline-delimited JSON sources passed to {@link
+ * SessionContext#registerJson(String, String, NdJsonReadOptions)} and {@link
+ * SessionContext#readJson(String, NdJsonReadOptions)}.
+ *
+ * <p>Mirrors a subset of DataFusion's {@code NdJsonReadOptions}. All setters return {@code this}
+ * for fluent chaining. Defaults match the Rust struct: {@code fileExtension = ".json"}, {@code
+ * fileCompressionType = UNCOMPRESSED}; {@code schemaInferMaxRecords} unset (the DataFusion default
+ * is used).
+ */
+public final class NdJsonReadOptions {
+
+  private String fileExtension = ".json";
+  private FileCompressionType fileCompressionType = FileCompressionType.UNCOMPRESSED;
+  private Long schemaInferMaxRecords;
+  private Schema schema;
+
+  public NdJsonReadOptions fileExtension(String ext) {
+    this.fileExtension = ext;
+    return this;
+  }
+
+  public NdJsonReadOptions fileCompressionType(FileCompressionType t) {
+    this.fileCompressionType = t;
+    return this;
+  }
+
+  public NdJsonReadOptions schemaInferMaxRecords(long n) {
+    if (n < 0) {
+      throw new IllegalArgumentException("schemaInferMaxRecords must be non-negative, got " + n);
+    }
+    this.schemaInferMaxRecords = n;
+    return this;
+  }
+
+  public NdJsonReadOptions schema(Schema schema) {
+    this.schema = schema;
+    return this;
+  }
+
+  byte[] toBytes() {
+    org.apache.datafusion.protobuf.NdJsonReadOptionsProto.Builder b =
+        org.apache.datafusion.protobuf.NdJsonReadOptionsProto.newBuilder()
+            .setFileExtension(fileExtension)
+            .setFileCompressionType(FileCompressionTypes.toProto(fileCompressionType));
+    if (schemaInferMaxRecords != null) {
+      b.setSchemaInferMaxRecords(schemaInferMaxRecords);
+    }
+    return b.build().toByteArray();
+  }
+
+  Schema schema() {
+    return schema;
+  }
+}

core/src/main/java/org/apache/datafusion/SessionContext.java

@@ -160,6 +160,67 @@ public DataFrame readCsv(String path, CsvReadOptions options) {
     return new DataFrame(dfHandle);
   }
 
+  public void registerJson(String name, String path) {
+    registerJson(name, path, new NdJsonReadOptions());
+  }
+
+  /**
+   * Register a newline-delimited JSON file (or directory of NDJSON files) as a table with the
+   * supplied {@link NdJsonReadOptions}.
+   *
+   * @throws RuntimeException if registration fails (path not found, schema inference error, etc.).
+   */
+  public void registerJson(String name, String path, NdJsonReadOptions options) {
+    if (nativeHandle == 0) {
+      throw new IllegalStateException("SessionContext is closed");
+    }
+    if (name == null) {
+      throw new IllegalArgumentException("registerJson name must be non-null");
+    }
+    if (path == null) {
+      throw new IllegalArgumentException("registerJson path must be non-null");
+    }
+    if (options == null) {
+      throw new IllegalArgumentException("registerJson options must be non-null");
+    }
+    registerJsonWithOptions(
+        nativeHandle,
+        name,
+        path,
+        options.toBytes(),
+        options.schema() != null ? serializeSchemaIpc(options.schema()) : null);
+  }
+
+  /** Read a newline-delimited JSON file as a {@link DataFrame} without registering it. */
+  public DataFrame readJson(String path) {
+    return readJson(path, new NdJsonReadOptions());
+  }
+
+  /**
+   * Read a newline-delimited JSON file as a {@link DataFrame} with the supplied {@link
+   * NdJsonReadOptions}.
+   *
+   * @throws RuntimeException if the read fails.
+   */
+  public DataFrame readJson(String path, NdJsonReadOptions options) {
+    if (nativeHandle == 0) {
+      throw new IllegalStateException("SessionContext is closed");
+    }
+    if (path == null) {
+      throw new IllegalArgumentException("readJson path must be non-null");
+    }
+    if (options == null) {
+      throw new IllegalArgumentException("readJson options must be non-null");
+    }
+    long dfHandle =
+        readJsonWithOptions(
+            nativeHandle,
+            path,
+            options.toBytes(),
+            options.schema() != null ? serializeSchemaIpc(options.schema()) : null);
+    return new DataFrame(dfHandle);
+  }
+
   public void registerParquet(String name, String path) {
     registerParquet(name, path, new ParquetReadOptions());
   }
@@ -247,5 +308,11 @@ private static native void registerCsvWithOptions(
   private static native long readCsvWithOptions(
       long handle, String path, byte[] optionsBytes, byte[] schemaIpcBytes);
 
+  private static native void registerJsonWithOptions(
+      long handle, String name, String path, byte[] optionsBytes, byte[] schemaIpcBytes);
+
+  private static native long readJsonWithOptions(
+      long handle, String path, byte[] optionsBytes, byte[] schemaIpcBytes);
+
   private static native void closeSessionContext(long handle);
 }

core/src/test/java/org/apache/datafusion/CsvReadOptionsTest.java

@@ -22,6 +22,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.util.List;
@@ -31,7 +32,6 @@
 import org.apache.arrow.vector.types.pojo.FieldType;
 import org.apache.arrow.vector.types.pojo.Schema;
 import org.apache.datafusion.protobuf.CsvReadOptionsProto;
-import org.apache.datafusion.protobuf.FileCompressionType;
 import org.junit.jupiter.api.Test;
 
 import com.google.protobuf.InvalidProtocolBufferException;
@@ -47,7 +47,8 @@ void defaultsRoundTripThroughProto() throws InvalidProtocolBufferException {
     assertEquals((int) '"', p.getQuote());
     assertEquals(".csv", p.getFileExtension());
     assertEquals(
-        FileCompressionType.FILE_COMPRESSION_TYPE_UNCOMPRESSED, p.getFileCompressionType());
+        org.apache.datafusion.protobuf.FileCompressionType.FILE_COMPRESSION_TYPE_UNCOMPRESSED,
+        p.getFileCompressionType());
 
     assertFalse(p.hasTerminator());
     assertFalse(p.hasEscape());
@@ -69,7 +70,7 @@ void fullyConfiguredRoundTripsThroughProto() throws InvalidProtocolBufferExcepti
             .newlinesInValues(true)
             .schemaInferMaxRecords(10L)
             .fileExtension(".tsv")
-            .fileCompressionType(CsvReadOptions.FileCompressionType.GZIP);
+            .fileCompressionType(FileCompressionType.GZIP);
 
     CsvReadOptionsProto p = CsvReadOptionsProto.parseFrom(opts.toBytes());
 
@@ -82,7 +83,9 @@ void fullyConfiguredRoundTripsThroughProto() throws InvalidProtocolBufferExcepti
     assertTrue(p.getNewlinesInValues());
     assertEquals(10L, p.getSchemaInferMaxRecords());
     assertEquals(".tsv", p.getFileExtension());
-    assertEquals(FileCompressionType.FILE_COMPRESSION_TYPE_GZIP, p.getFileCompressionType());
+    assertEquals(
+        org.apache.datafusion.protobuf.FileCompressionType.FILE_COMPRESSION_TYPE_GZIP,
+        p.getFileCompressionType());
   }
 
   @Test
@@ -96,7 +99,7 @@ void schemaIsHeldByReferenceAndNotInProto() {
 
   @Test
   void allCompressionTypesMapThroughProto() throws InvalidProtocolBufferException {
-    for (CsvReadOptions.FileCompressionType t : CsvReadOptions.FileCompressionType.values()) {
+    for (FileCompressionType t : FileCompressionType.values()) {
       CsvReadOptionsProto p =
           CsvReadOptionsProto.parseFrom(new CsvReadOptions().fileCompressionType(t).toBytes());
       assertEquals(
@@ -105,4 +108,14 @@ void allCompressionTypesMapThroughProto() throws InvalidProtocolBufferException
           "mismatch for " + t);
     }
   }
+
+  @Test
+  void schemaInferMaxRecordsRejectsNegative() {
+    // The proto wire field is uint64, so a negative long would be reinterpreted
+    // as a huge unsigned value on the Rust side and silently expand schema
+    // inference across the full dataset. Reject at the Java setter instead.
+    CsvReadOptions opts = new CsvReadOptions();
+    assertThrows(IllegalArgumentException.class, () -> opts.schemaInferMaxRecords(-1L));
+    assertThrows(IllegalArgumentException.class, () -> opts.schemaInferMaxRecords(Long.MIN_VALUE));
+  }
 }