Add Iceberg partition transforms: year/month/day/hour (#3903)

contributes to NVIDIA/spark-rapids#13384
contributes to NVIDIA/spark-rapids#13385
contributes to NVIDIA/spark-rapids#13386
contributes to NVIDIA/spark-rapids#13387

Add Iceberg partition transforms: year/month/day/hour.
CPU code is in `iceberg org.apache.iceberg.util.DateTimeUtil`,
[link](https://github.com/apache/iceberg/blob/apache-iceberg-1.6.1/api/src/main/java/org/apache/iceberg/util/DateTimeUtil.java)


### details
We can not use `ColumnView.year`: input MUST be timestamp type instead
of date type.

This PR adds the following four APIs:
```java
  /**
   * Calculates the difference in years between the epoch year (1970) and the
   * given date/timestamp column. E.g.: for date '1971-01-01', the result would be
   * 1: (1 year after epoch year)
   *
   * @param input The input date/timestamp column.
   * @return A column of type INT32 containing the year differences from epoch.
   */
  public static ColumnVector toYears(ColumnView input) 

  /**
   * Calculates the difference in months between the epoch month (1970-01) and the
   * given date/timestamp column. E.g.: for date '1971-02-01', the result would be
   * 13: (1 year and 1 month after epoch month)
   *
   * @param input The input date/timestamp column.
   * @return A column of type INT32 containing the month differences from epoch.
   */
  public static ColumnVector toMonths(ColumnView input) 

  /**
   * Calculates the difference in days between the epoch day (1970-01-01) and the
   * given date/timestamp column. E.g.: for date '1970-01-21', the result would be
   * 20: (20 days after epoch day)
   *
   * @param input The input date/timestamp column.
   * @return A column of type Date.
   */
  public static ColumnVector toDays(ColumnView input)

  /**
   * Calculates the difference in hours between the epoch hour
   * (1970-01-01T00:00:00) and the given timestamp column.
   * E.g.: for timestamp '1970-01-01 01:00:00', the result would be 1
   * (1 hour after epoch hour)
   *
   * @param timestamp The input timestamp column.
   * @return A column of type INT32 containing the hour differences from epoch.
   */
  public static ColumnVector toHours(ColumnView input) 
```

Signed-off-by: Chong Gao <res_life@163.com>

---------

Signed-off-by: Chong Gao <res_life@163.com>
Co-authored-by: Chong Gao <res_life@163.com>
Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

Author: 3 people

pom.xml

@@ -112,6 +112,7 @@
     <parquet.version>1.15.2</parquet.version>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <slf4j.version>1.7.30</slf4j.version>
+    <iceberg.version>1.6.1</iceberg.version>
     <submodule.check.skip>false</submodule.check.skip>
     <!-- This skips applying and unapplying patchs to CUDF. Be aware that if you 
     try to build the main jar without patching CUDF it could result in build
@@ -194,6 +195,12 @@
       <version>${hilbert.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+        <groupId>org.apache.iceberg</groupId>
+        <artifactId>iceberg-core</artifactId>
+        <version>${iceberg.version}</version>
+        <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <profiles>

src/main/cpp/CMakeLists.txt

@@ -198,6 +198,7 @@ add_library(
   src/HistogramJni.cpp
   src/HostTableJni.cpp
   src/HyperLogLogPlusPlusHostUDFJni.cpp
+  src/IcebergDateTimeUtilJni.cpp
   src/JSONUtilsJni.cpp
   src/KudoGpuSerializerJni.cpp
   src/ListSliceJni.cpp
@@ -235,6 +236,7 @@ add_library(
   src/hive_hash.cu
   src/hyper_log_log_plus_plus.cu
   src/hyper_log_log_plus_plus_host_udf.cu
+  src/iceberg_datetime_util.cu
   src/json_utils.cu
   src/list_slice.cu
   src/map.cu

src/main/cpp/src/IcebergDateTimeUtilJni.cpp

@@ -0,0 +1,84 @@
+/*
+ * Copyright (c) 2025, 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.
+ */
+
+#include "cudf_jni_apis.hpp"
+#include "iceberg_datetime_util.hpp"
+
+extern "C" {
+
+JNIEXPORT jlong JNICALL Java_com_nvidia_spark_rapids_jni_iceberg_IcebergDateTimeUtil_yearsFromEpoch(
+  JNIEnv* env, jclass, jlong input)
+{
+  JNI_NULL_CHECK(env, input, "input column is null", 0);
+
+  JNI_TRY
+  {
+    cudf::jni::auto_set_device(env);
+    auto const input_cv = reinterpret_cast<cudf::column_view const*>(input);
+    auto output         = spark_rapids_jni::years_from_epoch(*input_cv);
+    return reinterpret_cast<jlong>(output.release());
+  }
+  JNI_CATCH(env, 0);
+}
+
+JNIEXPORT jlong JNICALL
+Java_com_nvidia_spark_rapids_jni_iceberg_IcebergDateTimeUtil_monthsFromEpoch(JNIEnv* env,
+                                                                             jclass,
+                                                                             jlong input)
+{
+  JNI_NULL_CHECK(env, input, "input column is null", 0);
+
+  JNI_TRY
+  {
+    cudf::jni::auto_set_device(env);
+    auto const input_cv = reinterpret_cast<cudf::column_view const*>(input);
+    auto output         = spark_rapids_jni::months_from_epoch(*input_cv);
+    return reinterpret_cast<jlong>(output.release());
+  }
+  JNI_CATCH(env, 0);
+}
+
+JNIEXPORT jlong JNICALL Java_com_nvidia_spark_rapids_jni_iceberg_IcebergDateTimeUtil_daysFromEpoch(
+  JNIEnv* env, jclass, jlong input)
+{
+  JNI_NULL_CHECK(env, input, "input column is null", 0);
+
+  JNI_TRY
+  {
+    cudf::jni::auto_set_device(env);
+    auto const input_cv = reinterpret_cast<cudf::column_view const*>(input);
+    auto output         = spark_rapids_jni::days_from_epoch(*input_cv);
+    return reinterpret_cast<jlong>(output.release());
+  }
+  JNI_CATCH(env, 0);
+}
+
+JNIEXPORT jlong JNICALL Java_com_nvidia_spark_rapids_jni_iceberg_IcebergDateTimeUtil_hoursFromEpoch(
+  JNIEnv* env, jclass, jlong input)
+{
+  JNI_NULL_CHECK(env, input, "input column is null", 0);
+
+  JNI_TRY
+  {
+    cudf::jni::auto_set_device(env);
+    auto const input_cv = reinterpret_cast<cudf::column_view const*>(input);
+    auto output         = spark_rapids_jni::hours_from_epoch(*input_cv);
+    return reinterpret_cast<jlong>(output.release());
+  }
+  JNI_CATCH(env, 0);
+}
+
+}  // extern "C"

src/main/cpp/src/datetime_utils.cuh

@@ -16,6 +16,8 @@
 
 #pragma once
 
+#include "integer_utils.cuh"
+
 #include <cudf/lists/list_device_view.cuh>
 #include <cudf/lists/lists_column_device_view.cuh>
 #include <cudf/types.hpp>
@@ -230,7 +232,7 @@ struct date_time_utils {
   __device__ static int32_t get_year(int64_t seconds)
   {
     constexpr int64_t seconds_per_day = 86400L;
-    int64_t days                      = seconds / seconds_per_day;
+    int64_t days                      = integer_utils::floor_div(seconds, seconds_per_day);
     int32_t year, month, day;
     to_date(days, year, month, day);
     return year;