|
| 1 | +/* |
| 2 | + * Licensed to the Apache Software Foundation (ASF) under one |
| 3 | + * or more contributor license agreements. See the NOTICE file |
| 4 | + * distributed with this work for additional information |
| 5 | + * regarding copyright ownership. The ASF licenses this file |
| 6 | + * to you under the Apache License, Version 2.0 (the |
| 7 | + * "License"); you may not use this file except in compliance |
| 8 | + * with the License. You may obtain a copy of the License at |
| 9 | + * |
| 10 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 11 | + * |
| 12 | + * Unless required by applicable law or agreed to in writing, |
| 13 | + * software distributed under the License is distributed on an |
| 14 | + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 15 | + * KIND, either express or implied. See the License for the |
| 16 | + * specific language governing permissions and limitations |
| 17 | + * under the License. |
| 18 | + */ |
| 19 | + |
| 20 | +package org.apache.iceberg.mr.hive.udf; |
| 21 | + |
| 22 | +import java.nio.ByteBuffer; |
| 23 | +import java.nio.charset.StandardCharsets; |
| 24 | +import org.apache.hadoop.hive.ql.exec.Description; |
| 25 | +import org.apache.hadoop.hive.ql.exec.UDFArgumentException; |
| 26 | +import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException; |
| 27 | +import org.apache.hadoop.hive.ql.metadata.HiveException; |
| 28 | +import org.apache.hadoop.hive.ql.udf.generic.GenericUDF; |
| 29 | +import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector; |
| 30 | +import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector; |
| 31 | +import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory; |
| 32 | +import org.apache.hadoop.io.BytesWritable; |
| 33 | +import org.apache.iceberg.util.ZOrderByteUtils; |
| 34 | + |
| 35 | +/** |
| 36 | + * Hive UDF to compute the Z-order value of given input columns using Iceberg's ZOrderByteUtils. |
| 37 | + * Supports various primitive types and converts inputs into interleaved binary representation. |
| 38 | + */ |
| 39 | +@Description(name = "iceberg_zorder", |
| 40 | + value = "_FUNC_(value) - " + |
| 41 | + "Returns the z-value calculated by Iceberg ZOrderByteUtils class") |
| 42 | +public class GenericUDFIcebergZorder extends GenericUDF { |
| 43 | + private PrimitiveObjectInspector[] argOIs; |
| 44 | + // For variable-length types (e.g., strings), how many bytes contribute to z-order |
| 45 | + private final int varLengthContribution = 8; |
| 46 | + private transient ByteBuffer[] reUseBuffer; |
| 47 | + private static final int MAX_OUTPUT_SIZE = Integer.MAX_VALUE; |
| 48 | + |
| 49 | + /** |
| 50 | + * Initializes the UDF, validating argument types are primitives and preparing buffers. |
| 51 | + */ |
| 52 | + @Override |
| 53 | + public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException { |
| 54 | + if (arguments.length < 2) { |
| 55 | + throw new UDFArgumentException("iceberg_zorder requires at least 2 arguments"); |
| 56 | + } |
| 57 | + argOIs = new PrimitiveObjectInspector[arguments.length]; |
| 58 | + reUseBuffer = new ByteBuffer[arguments.length]; |
| 59 | + for (int i = 0; i < arguments.length; i++) { |
| 60 | + if (!(arguments[i] instanceof PrimitiveObjectInspector poi)) { |
| 61 | + throw new UDFArgumentTypeException(i, "Only primitive types supported for z-order"); |
| 62 | + } |
| 63 | + argOIs[i] = poi; |
| 64 | + } |
| 65 | + return PrimitiveObjectInspectorFactory.writableBinaryObjectInspector; |
| 66 | + } |
| 67 | + |
| 68 | + /** |
| 69 | + * Evaluates the UDF by converting input values to ordered bytes, interleaving them, |
| 70 | + * and returning the resulting Z-order binary value. |
| 71 | + */ |
| 72 | + @Override |
| 73 | + public Object evaluate(DeferredObject[] arguments) throws HiveException { |
| 74 | + byte[][] inputs = new byte[arguments.length][]; |
| 75 | + int totalLength = 0; |
| 76 | + |
| 77 | + for (int i = 0; i < arguments.length; i++) { |
| 78 | + byte[] orderedBytes = convertToOrderedBytes(arguments[i].get(), argOIs[i], i); |
| 79 | + inputs[i] = orderedBytes; |
| 80 | + totalLength += orderedBytes.length; |
| 81 | + } |
| 82 | + |
| 83 | + int outputLength = Math.min(totalLength, MAX_OUTPUT_SIZE); |
| 84 | + ByteBuffer buffer = ByteBuffer.allocate(outputLength); |
| 85 | + |
| 86 | + byte[] interleaved = ZOrderByteUtils.interleaveBits(inputs, outputLength, buffer); |
| 87 | + return new BytesWritable(interleaved); |
| 88 | + } |
| 89 | + |
| 90 | + @Override |
| 91 | + public String getDisplayString(String[] children) { |
| 92 | + return "iceberg_zorder(" + String.join(", ", children) + ")"; |
| 93 | + } |
| 94 | + |
| 95 | + /** |
| 96 | + * Converts a single input value to its ordered byte representation based on type. |
| 97 | + * @return fixed-length byte arrays to be used in interleaving. |
| 98 | + */ |
| 99 | + private byte[] convertToOrderedBytes(Object value, PrimitiveObjectInspector oi, |
| 100 | + int position) throws HiveException { |
| 101 | + if (value == null) { |
| 102 | + // For NULL values, we have primitive buffer size of 8 with values of 0 |
| 103 | + return ByteBuffer.wrap(new byte[ZOrderByteUtils.PRIMITIVE_BUFFER_SIZE]).array(); |
| 104 | + } |
| 105 | + |
| 106 | + if (reUseBuffer[position] == null) { |
| 107 | + reUseBuffer[position] = ByteBuffer.allocate(ZOrderByteUtils.PRIMITIVE_BUFFER_SIZE); |
| 108 | + } |
| 109 | + switch (oi.getPrimitiveCategory()) { |
| 110 | + case BOOLEAN: |
| 111 | + boolean boolValue = (Boolean) oi.getPrimitiveJavaObject(value); |
| 112 | + return ZOrderByteUtils.intToOrderedBytes(boolValue ? 1 : 0, reUseBuffer[position]).array(); |
| 113 | + |
| 114 | + case BYTE: |
| 115 | + byte byteValue = (Byte) oi.getPrimitiveJavaObject(value); |
| 116 | + return ZOrderByteUtils.tinyintToOrderedBytes(byteValue, reUseBuffer[position]).array(); |
| 117 | + |
| 118 | + case SHORT: |
| 119 | + short shortValue = (Short) oi.getPrimitiveJavaObject(value); |
| 120 | + return ZOrderByteUtils.shortToOrderedBytes(shortValue, reUseBuffer[position]).array(); |
| 121 | + |
| 122 | + case INT: |
| 123 | + int intValue = (Integer) oi.getPrimitiveJavaObject(value); |
| 124 | + return ZOrderByteUtils.intToOrderedBytes(intValue, reUseBuffer[position]).array(); |
| 125 | + |
| 126 | + case LONG: |
| 127 | + long longValue = (Long) oi.getPrimitiveJavaObject(value); |
| 128 | + return ZOrderByteUtils.longToOrderedBytes(longValue, reUseBuffer[position]).array(); |
| 129 | + |
| 130 | + case FLOAT: |
| 131 | + float floatValue = (Float) oi.getPrimitiveJavaObject(value); |
| 132 | + return ZOrderByteUtils.floatToOrderedBytes(floatValue, reUseBuffer[position]).array(); |
| 133 | + |
| 134 | + case DOUBLE: |
| 135 | + double doubleValue = (Double) oi.getPrimitiveJavaObject(value); |
| 136 | + return ZOrderByteUtils.doubleToOrderedBytes(doubleValue, reUseBuffer[position]).array(); |
| 137 | + |
| 138 | + case DATE: |
| 139 | + // Get data in epoch seconds and convert it to long |
| 140 | + Object dateValue = oi.getPrimitiveJavaObject(value); |
| 141 | + long dateInSeconds; |
| 142 | + if (dateValue instanceof java.sql.Date dd) { |
| 143 | + dateInSeconds = dd.getTime() / 1000L; |
| 144 | + } else if (dateValue instanceof org.apache.hadoop.hive.common.type.Date dd) { |
| 145 | + dateInSeconds = dd.toEpochSecond(); |
| 146 | + } else { |
| 147 | + throw new HiveException("Unsupported DATE backing type: " + dateValue.getClass()); |
| 148 | + } |
| 149 | + return ZOrderByteUtils.longToOrderedBytes(dateInSeconds, reUseBuffer[position]).array(); |
| 150 | + |
| 151 | + case TIMESTAMP: |
| 152 | + Object tsValue = oi.getPrimitiveJavaObject(value); |
| 153 | + long tsInSeconds; |
| 154 | + if (tsValue instanceof org.apache.hadoop.hive.common.type.Timestamp ts) { |
| 155 | + tsInSeconds = ts.toEpochSecond(); |
| 156 | + } else if (tsValue instanceof java.sql.Timestamp ts) { |
| 157 | + tsInSeconds = ts.getTime() / 1000L; |
| 158 | + } else { |
| 159 | + throw new HiveException("Unsupported TIMESTAMP backing type: " + tsValue.getClass()); |
| 160 | + } |
| 161 | + return ZOrderByteUtils.longToOrderedBytes(tsInSeconds, reUseBuffer[position]).array(); |
| 162 | + |
| 163 | + case CHAR: |
| 164 | + case VARCHAR: |
| 165 | + case STRING: |
| 166 | + String strVal = String.valueOf(oi.getPrimitiveJavaObject(value)); |
| 167 | + return ZOrderByteUtils.stringToOrderedBytes(strVal, varLengthContribution, |
| 168 | + reUseBuffer[position], StandardCharsets.UTF_8.newEncoder()).array(); |
| 169 | + |
| 170 | + default: |
| 171 | + throw new HiveException("Unsupported type in z-order: " + oi.getPrimitiveCategory()); |
| 172 | + } |
| 173 | + } |
| 174 | +} |
0 commit comments