Introduce kudo writer. (#2559)

* Introduce kudo writer

Signed-off-by: liurenjie1024 <[email protected]>

src/main/java/com/nvidia/spark/rapids/jni/Preconditions.java

@@ -21,7 +21,7 @@
 /**
  * This class contains utility methods for checking preconditions.
  */
-class Preconditions {
+public class Preconditions {
     /**
      * Check if the condition is true, otherwise throw an IllegalStateException with the given message.
      */
@@ -39,4 +39,18 @@ public static void ensure(boolean condition, Supplier<String> messageSupplier) {
             throw new IllegalStateException(messageSupplier.get());
         }
     }
+
+    /**
+     * Check if the value is non-negative, otherwise throw an IllegalArgumentException with the given message.
+     * @param value the value to check
+     * @param name the name of the value
+     * @return the value if it is non-negative
+     * @throws IllegalArgumentException if the value is negative
+     */
+    public static int ensureNonNegative(int value, String name) {
+        if (value < 0) {
+            throw new IllegalArgumentException(name + " must be non-negative, but was " + value);
+        }
+        return value;
+    }
 }

src/main/java/com/nvidia/spark/rapids/jni/kudo/DataOutputStreamWriter.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright (c) 2024, NVIDIA CORPORATION.
+ *
+ * Licensed 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 com.nvidia.spark.rapids.jni.kudo;
+
+import ai.rapids.cudf.HostMemoryBuffer;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+/**
+ * Visible for testing
+ */
+class DataOutputStreamWriter extends DataWriter {
+  private final byte[] arrayBuffer = new byte[1024 * 128];
+  private final DataOutputStream dout;
+
+  public DataOutputStreamWriter(DataOutputStream dout) {
+    this.dout = dout;
+  }
+
+  @Override
+  public void writeInt(int i) throws IOException {
+    dout.writeInt(i);
+  }
+
+  @Override
+  public void copyDataFrom(HostMemoryBuffer src, long srcOffset, long len) throws IOException {
+    long dataLeft = len;
+    while (dataLeft > 0) {
+      int amountToCopy = (int) Math.min(arrayBuffer.length, dataLeft);
+      src.getBytes(arrayBuffer, 0, srcOffset, amountToCopy);
+      dout.write(arrayBuffer, 0, amountToCopy);
+      srcOffset += amountToCopy;
+      dataLeft -= amountToCopy;
+    }
+  }
+
+  @Override
+  public void flush() throws IOException {
+    dout.flush();
+  }
+
+  @Override
+  public void write(byte[] arr, int offset, int length) throws IOException {
+    dout.write(arr, offset, length);
+  }
+}

src/main/java/com/nvidia/spark/rapids/jni/kudo/DataWriter.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright (c) 2024, NVIDIA CORPORATION.
+ *
+ * Licensed 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 com.nvidia.spark.rapids.jni.kudo;
+
+import ai.rapids.cudf.HostMemoryBuffer;
+
+import java.io.IOException;
+
+/**
+ * Visible for testing
+ */
+abstract class DataWriter {
+
+  public abstract void writeInt(int i) throws IOException;
+
+  /**
+   * Copy data from src starting at srcOffset and going for len bytes.
+   *
+   * @param src       where to copy from.
+   * @param srcOffset offset to start at.
+   * @param len       amount to copy.
+   */
+  public abstract void copyDataFrom(HostMemoryBuffer src, long srcOffset, long len) throws IOException;
+
+  public void flush() throws IOException {
+    // NOOP by default
+  }
+
+  public abstract void write(byte[] arr, int offset, int length) throws IOException;
+}