| /** |
| * 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.tajo.engine.planner; |
| |
| import com.google.common.base.Preconditions; |
| import com.google.common.collect.Lists; |
| import org.apache.tajo.catalog.Column; |
| import org.apache.tajo.catalog.SortSpec; |
| import org.apache.tajo.datum.Datum; |
| import org.apache.tajo.datum.DatumFactory; |
| import org.apache.tajo.datum.NullDatum; |
| import org.apache.tajo.engine.exception.RangeOverflowException; |
| import org.apache.tajo.storage.Tuple; |
| import org.apache.tajo.storage.TupleRange; |
| import org.apache.tajo.storage.VTuple; |
| import org.apache.tajo.util.Bytes; |
| |
| import java.math.BigDecimal; |
| import java.math.RoundingMode; |
| import java.util.List; |
| |
| |
| public class UniformRangePartition extends RangePartitionAlgorithm { |
| private int variableId; |
| private BigDecimal[] cardForEachDigit; |
| private BigDecimal[] colCards; |
| |
| /** |
| * |
| * @param totalRange |
| * @param sortSpecs The description of sort keys |
| * @param inclusive true if the end of the range is inclusive |
| */ |
| public UniformRangePartition(TupleRange totalRange, SortSpec[] sortSpecs, boolean inclusive) { |
| super(sortSpecs, totalRange, inclusive); |
| colCards = new BigDecimal[sortSpecs.length]; |
| for (int i = 0; i < sortSpecs.length; i++) { |
| colCards[i] = computeCardinality(sortSpecs[i].getSortKey().getDataType(), totalRange.getStart().get(i), |
| totalRange.getEnd().get(i), inclusive, sortSpecs[i].isAscending()); |
| } |
| |
| cardForEachDigit = new BigDecimal[colCards.length]; |
| for (int i = 0; i < colCards.length ; i++) { |
| if (i == 0) { |
| cardForEachDigit[i] = colCards[i]; |
| } else { |
| cardForEachDigit[i] = cardForEachDigit[i - 1].multiply(colCards[i]); |
| } |
| } |
| } |
| |
| public UniformRangePartition(TupleRange range, SortSpec [] sortSpecs) { |
| this(range, sortSpecs, true); |
| } |
| |
| @Override |
| public TupleRange[] partition(int partNum) { |
| Preconditions.checkArgument(partNum > 0, |
| "The number of partitions must be positive, but the given number: " |
| + partNum); |
| Preconditions.checkArgument(totalCard.compareTo(new BigDecimal(partNum)) >= 0, |
| "the number of partition cannot exceed total cardinality (" + totalCard + ")"); |
| |
| int varId; |
| for (varId = 0; varId < cardForEachDigit.length; varId++) { |
| if (cardForEachDigit[varId].compareTo(new BigDecimal(partNum)) >= 0) |
| break; |
| } |
| this.variableId = varId; |
| |
| BigDecimal [] reverseCardsForDigit = new BigDecimal[variableId+1]; |
| for (int i = variableId; i >= 0; i--) { |
| if (i == variableId) { |
| reverseCardsForDigit[i] = colCards[i]; |
| } else { |
| reverseCardsForDigit[i] = reverseCardsForDigit[i+1].multiply(colCards[i]); |
| } |
| } |
| |
| List<TupleRange> ranges = Lists.newArrayList(); |
| BigDecimal term = reverseCardsForDigit[0].divide( |
| new BigDecimal(partNum), RoundingMode.CEILING); |
| BigDecimal reminder = reverseCardsForDigit[0]; |
| Tuple last = range.getStart(); |
| while(reminder.compareTo(new BigDecimal(0)) > 0) { |
| if (reminder.compareTo(term) <= 0) { // final one is inclusive |
| ranges.add(new TupleRange(sortSpecs, last, range.getEnd())); |
| } else { |
| Tuple next = increment(last, term.longValue(), variableId); |
| ranges.add(new TupleRange(sortSpecs, last, next)); |
| } |
| last = ranges.get(ranges.size() - 1).getEnd(); |
| reminder = reminder.subtract(term); |
| } |
| |
| return ranges.toArray(new TupleRange[ranges.size()]); |
| } |
| |
| /** |
| * Check whether an overflow occurs or not. |
| * |
| * @param colId The column id to be checked |
| * @param last |
| * @param inc |
| * @param sortSpecs |
| * @return |
| */ |
| public boolean isOverflow(int colId, Datum last, BigDecimal inc, SortSpec [] sortSpecs) { |
| Column column = sortSpecs[colId].getSortKey(); |
| BigDecimal candidate; |
| boolean overflow = false; |
| |
| switch (column.getDataType().getType()) { |
| case BIT: { |
| if (sortSpecs[colId].isAscending()) { |
| candidate = inc.add(new BigDecimal(last.asByte())); |
| return new BigDecimal(range.getEnd().get(colId).asByte()).compareTo(candidate) < 0; |
| } else { |
| candidate = new BigDecimal(last.asByte()).subtract(inc); |
| return candidate.compareTo(new BigDecimal(range.getEnd().get(colId).asByte())) < 0; |
| } |
| } |
| case CHAR: { |
| if (sortSpecs[colId].isAscending()) { |
| candidate = inc.add(new BigDecimal((int)last.asChar())); |
| return new BigDecimal((int)range.getEnd().get(colId).asChar()).compareTo(candidate) < 0; |
| } else { |
| candidate = new BigDecimal((int)last.asChar()).subtract(inc); |
| return candidate.compareTo(new BigDecimal((int)range.getEnd().get(colId).asChar())) < 0; |
| } |
| } |
| case INT2: { |
| if (sortSpecs[colId].isAscending()) { |
| candidate = inc.add(new BigDecimal(last.asInt2())); |
| return new BigDecimal(range.getEnd().get(colId).asInt2()).compareTo(candidate) < 0; |
| } else { |
| candidate = new BigDecimal(last.asInt2()).subtract(inc); |
| return candidate.compareTo(new BigDecimal(range.getEnd().get(colId).asInt2())) < 0; |
| } |
| } |
| case DATE: |
| case INT4: { |
| if (sortSpecs[colId].isAscending()) { |
| candidate = inc.add(new BigDecimal(last.asInt4())); |
| return new BigDecimal(range.getEnd().get(colId).asInt4()).compareTo(candidate) < 0; |
| } else { |
| candidate = new BigDecimal(last.asInt4()).subtract(inc); |
| return candidate.compareTo(new BigDecimal(range.getEnd().get(colId).asInt4())) < 0; |
| } |
| } |
| case TIME: |
| case TIMESTAMP: |
| case INT8: { |
| if (sortSpecs[colId].isAscending()) { |
| candidate = inc.add(new BigDecimal(last.asInt8())); |
| return new BigDecimal(range.getEnd().get(colId).asInt8()).compareTo(candidate) < 0; |
| } else { |
| candidate = new BigDecimal(last.asInt8()).subtract(inc); |
| return candidate.compareTo(new BigDecimal(range.getEnd().get(colId).asInt8())) < 0; |
| } |
| } |
| case FLOAT4: { |
| if (sortSpecs[colId].isAscending()) { |
| candidate = inc.add(new BigDecimal(last.asFloat4())); |
| return new BigDecimal(range.getEnd().get(colId).asFloat4()).compareTo(candidate) < 0; |
| } else { |
| candidate = new BigDecimal(last.asFloat4()).subtract(inc); |
| return candidate.compareTo(new BigDecimal(range.getEnd().get(colId).asFloat4())) < 0; |
| } |
| } |
| case FLOAT8: { |
| if (sortSpecs[colId].isAscending()) { |
| candidate = inc.add(new BigDecimal(last.asFloat8())); |
| return new BigDecimal(range.getEnd().get(colId).asFloat8()).compareTo(candidate) < 0; |
| } else { |
| candidate = new BigDecimal(last.asFloat8()).subtract(inc); |
| return candidate.compareTo(new BigDecimal(range.getEnd().get(colId).asFloat8())) < 0; |
| } |
| |
| } |
| case TEXT: { |
| if (sortSpecs[colId].isAscending()) { |
| candidate = inc.add(new BigDecimal((int)(last instanceof NullDatum ? '0' : last.asChars().charAt(0)))); |
| return new BigDecimal(range.getEnd().get(colId).asChars().charAt(0)).compareTo(candidate) < 0; |
| } else { |
| candidate = new BigDecimal((int)(last.asChars().charAt(0))).subtract(inc); |
| return candidate.compareTo(new BigDecimal(range.getEnd().get(colId).asChars().charAt(0))) < 0; |
| } |
| } |
| case INET4: { |
| int candidateIntVal; |
| byte[] candidateBytesVal = new byte[4]; |
| if (sortSpecs[colId].isAscending()) { |
| candidateIntVal = inc.intValue() + last.asInt4(); |
| if (candidateIntVal - inc.intValue() != last.asInt4()) { |
| return true; |
| } |
| Bytes.putInt(candidateBytesVal, 0, candidateIntVal); |
| return Bytes.compareTo(range.getEnd().get(colId).asByteArray(), candidateBytesVal) < 0; |
| } else { |
| candidateIntVal = last.asInt4() - inc.intValue(); |
| if (candidateIntVal + inc.intValue() != last.asInt4()) { |
| return true; |
| } |
| Bytes.putInt(candidateBytesVal, 0, candidateIntVal); |
| return Bytes.compareTo(candidateBytesVal, range.getEnd().get(colId).asByteArray()) < 0; |
| } |
| } |
| } |
| return overflow; |
| } |
| |
| public long incrementAndGetReminder(int colId, Datum last, long inc) { |
| Column column = sortSpecs[colId].getSortKey(); |
| long reminder = 0; |
| switch (column.getDataType().getType()) { |
| case BIT: { |
| long candidate = last.asByte() + inc; |
| byte end = range.getEnd().get(colId).asByte(); |
| reminder = candidate - end; |
| break; |
| } |
| case CHAR: { |
| long candidate = last.asChar() + inc; |
| char end = range.getEnd().get(colId).asChar(); |
| reminder = candidate - end; |
| break; |
| } |
| case DATE: |
| case INT4: { |
| int candidate = (int) (last.asInt4() + inc); |
| int end = range.getEnd().get(colId).asInt4(); |
| reminder = candidate - end; |
| break; |
| } |
| case TIME: |
| case TIMESTAMP: |
| case INT8: |
| case INET4: { |
| long candidate = last.asInt8() + inc; |
| long end = range.getEnd().get(colId).asInt8(); |
| reminder = candidate - end; |
| break; |
| } |
| case FLOAT4: { |
| float candidate = last.asFloat4() + inc; |
| float end = range.getEnd().get(colId).asFloat4(); |
| reminder = (long) (candidate - end); |
| break; |
| } |
| case FLOAT8: { |
| double candidate = last.asFloat8() + inc; |
| double end = range.getEnd().get(colId).asFloat8(); |
| reminder = (long) Math.ceil(candidate - end); |
| break; |
| } |
| case TEXT: { |
| char candidate = ((char)(last.asChars().charAt(0) + inc)); |
| char end = range.getEnd().get(colId).asChars().charAt(0); |
| reminder = (char) (candidate - end); |
| break; |
| } |
| } |
| |
| // including zero |
| return reminder - 1; |
| } |
| |
| /** |
| * |
| * @param last |
| * @param inc |
| * @return |
| */ |
| public Tuple increment(final Tuple last, final long inc, final int baseDigit) { |
| BigDecimal [] incs = new BigDecimal[last.size()]; |
| boolean [] overflowFlag = new boolean[last.size()]; |
| BigDecimal [] result; |
| BigDecimal value = new BigDecimal(inc); |
| |
| BigDecimal [] reverseCardsForDigit = new BigDecimal[baseDigit + 1]; |
| for (int i = baseDigit; i >= 0; i--) { |
| if (i == baseDigit) { |
| reverseCardsForDigit[i] = colCards[i]; |
| } else { |
| reverseCardsForDigit[i] = reverseCardsForDigit[i+1].multiply(colCards[i]); |
| } |
| } |
| |
| for (int i = 0; i < baseDigit; i++) { |
| result = value.divideAndRemainder(reverseCardsForDigit[i + 1]); |
| incs[i] = result[0]; |
| value = result[1]; |
| } |
| int finalId = baseDigit; |
| incs[finalId] = value; |
| for (int i = finalId; i >= 0; i--) { |
| if (isOverflow(i, last.get(i), incs[i], sortSpecs)) { |
| if (i == 0) { |
| throw new RangeOverflowException(range, last, incs[i].longValue()); |
| } |
| long rem = incrementAndGetReminder(i, last.get(i), value.longValue()); |
| incs[i] = new BigDecimal(rem); |
| incs[i - 1] = incs[i-1].add(new BigDecimal(1)); |
| overflowFlag[i] = true; |
| } else { |
| if (i > 0) { |
| incs[i] = value; |
| break; |
| } |
| } |
| } |
| |
| for (int i = 0; i < incs.length; i++) { |
| if (incs[i] == null) { |
| incs[i] = new BigDecimal(0); |
| } |
| } |
| |
| Tuple end = new VTuple(sortSpecs.length); |
| Column column; |
| for (int i = 0; i < last.size(); i++) { |
| column = sortSpecs[i].getSortKey(); |
| switch (column.getDataType().getType()) { |
| case CHAR: |
| if (overflowFlag[i]) { |
| end.put(i, DatumFactory.createChar((char) (range.getStart().get(i).asChar() + incs[i].longValue()))); |
| } else { |
| end.put(i, DatumFactory.createChar((char) (last.get(i).asChar() + incs[i].longValue()))); |
| } |
| break; |
| case BIT: |
| if (overflowFlag[i]) { |
| end.put(i, DatumFactory.createBit( |
| (byte) (range.getStart().get(i).asByte() + incs[i].longValue()))); |
| } else { |
| end.put(i, DatumFactory.createBit((byte) (last.get(i).asByte() + incs[i].longValue()))); |
| } |
| break; |
| case INT2: |
| if (overflowFlag[i]) { |
| end.put(i, DatumFactory.createInt2( |
| (short) (range.getStart().get(i).asInt2() + incs[i].longValue()))); |
| } else { |
| end.put(i, DatumFactory.createInt2((short) (last.get(i).asInt2() + incs[i].longValue()))); |
| } |
| break; |
| case INT4: |
| if (overflowFlag[i]) { |
| end.put(i, DatumFactory.createInt4( |
| (int) (range.getStart().get(i).asInt4() + incs[i].longValue()))); |
| } else { |
| if (sortSpecs[i].isAscending()) { |
| end.put(i, DatumFactory.createInt4((int) (last.get(i).asInt4() + incs[i].longValue()))); |
| } else { |
| end.put(i, DatumFactory.createInt4((int) (last.get(i).asInt4() - incs[i].longValue()))); |
| } |
| } |
| break; |
| case INT8: |
| if (overflowFlag[i]) { |
| end.put(i, DatumFactory.createInt8( |
| range.getStart().get(i).asInt8() + incs[i].longValue())); |
| } else { |
| end.put(i, DatumFactory.createInt8(last.get(i).asInt8() + incs[i].longValue())); |
| } |
| break; |
| case FLOAT4: |
| if (overflowFlag[i]) { |
| end.put(i, DatumFactory.createFloat4( |
| range.getStart().get(i).asFloat4() + incs[i].longValue())); |
| } else { |
| end.put(i, DatumFactory.createFloat4(last.get(i).asFloat4() + incs[i].longValue())); |
| } |
| break; |
| case FLOAT8: |
| if (overflowFlag[i]) { |
| end.put(i, DatumFactory.createFloat8( |
| range.getStart().get(i).asFloat8() + incs[i].longValue())); |
| } else { |
| end.put(i, DatumFactory.createFloat8(last.get(i).asFloat8() + incs[i].longValue())); |
| } |
| break; |
| case TEXT: |
| if (overflowFlag[i]) { |
| end.put(i, DatumFactory.createText(((char) (range.getStart().get(i).asChars().charAt(0) |
| + incs[i].longValue())) + "")); |
| } else { |
| end.put(i, DatumFactory.createText( |
| ((char) ((last.get(i) instanceof NullDatum ? '0': last.get(i).asChars().charAt(0)) + incs[i].longValue())) + "")); |
| } |
| break; |
| case DATE: |
| if (overflowFlag[i]) { |
| end.put(i, DatumFactory.createDate((int) (range.getStart().get(i).asInt4() + incs[i].longValue()))); |
| } else { |
| end.put(i, DatumFactory.createDate((int) (last.get(i).asInt4() + incs[i].longValue()))); |
| } |
| break; |
| case TIME: |
| if (overflowFlag[i]) { |
| end.put(i, DatumFactory.createTime(range.getStart().get(i).asInt8() + incs[i].longValue())); |
| } else { |
| end.put(i, DatumFactory.createTime(last.get(i).asInt8() + incs[i].longValue())); |
| } |
| break; |
| case TIMESTAMP: |
| if (overflowFlag[i]) { |
| end.put(i, DatumFactory.createTimeStampFromMillis( |
| range.getStart().get(i).asInt8() + incs[i].longValue())); |
| } else { |
| end.put(i, DatumFactory.createTimeStampFromMillis(last.get(i).asInt8() + incs[i].longValue())); |
| } |
| break; |
| case INET4: |
| byte[] ipBytes; |
| if (overflowFlag[i]) { |
| ipBytes = range.getStart().get(i).asByteArray(); |
| assert ipBytes.length == 4; |
| end.put(i, DatumFactory.createInet4(ipBytes)); |
| } else { |
| int lastVal = last.get(i).asInt4() + incs[i].intValue(); |
| ipBytes = new byte[4]; |
| Bytes.putInt(ipBytes, 0, lastVal); |
| end.put(i, DatumFactory.createInet4(ipBytes)); |
| } |
| break; |
| default: |
| throw new UnsupportedOperationException(column.getDataType() + " is not supported yet"); |
| } |
| } |
| |
| return end; |
| } |
| } |
