| /* |
| * 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.phoenix.compile; |
| |
| import static com.google.common.base.Preconditions.checkArgument; |
| import static com.google.common.collect.Lists.newArrayListWithCapacity; |
| |
| import java.sql.ParameterMetaData; |
| import java.sql.ResultSet; |
| import java.sql.SQLException; |
| import java.sql.Timestamp; |
| import java.util.Arrays; |
| import java.util.BitSet; |
| import java.util.Collections; |
| import java.util.LinkedHashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| |
| import com.google.common.collect.ImmutableList; |
| import org.apache.hadoop.hbase.Cell; |
| import org.apache.hadoop.hbase.HRegionLocation; |
| import org.apache.hadoop.hbase.client.Scan; |
| import org.apache.hadoop.hbase.io.ImmutableBytesWritable; |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.hadoop.hbase.util.Pair; |
| import org.apache.phoenix.cache.ServerCacheClient; |
| import org.apache.phoenix.compile.GroupByCompiler.GroupBy; |
| import org.apache.phoenix.compile.OrderByCompiler.OrderBy; |
| import org.apache.phoenix.coprocessor.BaseScannerRegionObserver; |
| import org.apache.phoenix.coprocessor.MetaDataProtocol; |
| import org.apache.phoenix.coprocessor.UngroupedAggregateRegionObserver; |
| import org.apache.phoenix.exception.SQLExceptionCode; |
| import org.apache.phoenix.exception.SQLExceptionInfo; |
| import org.apache.phoenix.execute.AggregatePlan; |
| import org.apache.phoenix.execute.MutationState; |
| import org.apache.phoenix.execute.MutationState.MultiRowMutationState; |
| import org.apache.phoenix.execute.MutationState.RowMutationState; |
| import org.apache.phoenix.execute.MutationState.RowTimestampColInfo; |
| import org.apache.phoenix.expression.Determinism; |
| import org.apache.phoenix.expression.Expression; |
| import org.apache.phoenix.expression.LiteralExpression; |
| import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr; |
| import org.apache.phoenix.index.IndexMaintainer; |
| import org.apache.phoenix.index.PhoenixIndexBuilder; |
| import org.apache.phoenix.index.PhoenixIndexCodec; |
| import org.apache.phoenix.iterate.ResultIterator; |
| import org.apache.phoenix.jdbc.PhoenixConnection; |
| import org.apache.phoenix.jdbc.PhoenixResultSet; |
| import org.apache.phoenix.jdbc.PhoenixStatement; |
| import org.apache.phoenix.jdbc.PhoenixStatement.Operation; |
| import org.apache.phoenix.optimize.QueryOptimizer; |
| import org.apache.phoenix.parse.AliasedNode; |
| import org.apache.phoenix.parse.BindParseNode; |
| import org.apache.phoenix.parse.ColumnName; |
| import org.apache.phoenix.parse.HintNode; |
| import org.apache.phoenix.parse.HintNode.Hint; |
| import org.apache.phoenix.parse.LiteralParseNode; |
| import org.apache.phoenix.parse.NamedTableNode; |
| import org.apache.phoenix.parse.ParseNode; |
| import org.apache.phoenix.parse.SelectStatement; |
| import org.apache.phoenix.parse.SequenceValueParseNode; |
| import org.apache.phoenix.parse.UpsertStatement; |
| import org.apache.phoenix.query.ConnectionQueryServices; |
| import org.apache.phoenix.query.QueryConstants; |
| import org.apache.phoenix.query.QueryServices; |
| import org.apache.phoenix.query.QueryServicesOptions; |
| import org.apache.phoenix.schema.ColumnRef; |
| import org.apache.phoenix.schema.ConstraintViolationException; |
| import org.apache.phoenix.schema.DelegateColumn; |
| import org.apache.phoenix.schema.IllegalDataException; |
| import org.apache.phoenix.schema.MetaDataClient; |
| import org.apache.phoenix.schema.PColumn; |
| import org.apache.phoenix.schema.PColumnImpl; |
| import org.apache.phoenix.schema.PName; |
| import org.apache.phoenix.schema.PNameFactory; |
| import org.apache.phoenix.schema.PTable; |
| import org.apache.phoenix.schema.PTable.IndexType; |
| import org.apache.phoenix.schema.PTable.ViewType; |
| import org.apache.phoenix.schema.PTableImpl; |
| import org.apache.phoenix.schema.PTableKey; |
| import org.apache.phoenix.schema.PTableType; |
| import org.apache.phoenix.schema.ReadOnlyTableException; |
| import org.apache.phoenix.schema.SortOrder; |
| import org.apache.phoenix.schema.TableRef; |
| import org.apache.phoenix.schema.TypeMismatchException; |
| import org.apache.phoenix.schema.UpsertColumnsValuesMismatchException; |
| import org.apache.phoenix.schema.tuple.Tuple; |
| import org.apache.phoenix.schema.types.PDataType; |
| import org.apache.phoenix.schema.types.PLong; |
| import org.apache.phoenix.schema.types.PSmallint; |
| import org.apache.phoenix.schema.types.PTimestamp; |
| import org.apache.phoenix.schema.types.PUnsignedLong; |
| import org.apache.phoenix.schema.types.PVarbinary; |
| import org.apache.phoenix.util.ByteUtil; |
| import org.apache.phoenix.util.ExpressionUtil; |
| import org.apache.phoenix.util.IndexUtil; |
| import org.apache.phoenix.util.ScanUtil; |
| import org.apache.phoenix.util.SchemaUtil; |
| |
| import com.google.common.collect.Lists; |
| import com.google.common.collect.Maps; |
| import com.google.common.collect.Sets; |
| |
| public class UpsertCompiler { |
| |
| private static void setValues(byte[][] values, int[] pkSlotIndex, int[] columnIndexes, |
| PTable table, MultiRowMutationState mutation, |
| PhoenixStatement statement, boolean useServerTimestamp, IndexMaintainer maintainer, |
| byte[][] viewConstants, byte[] onDupKeyBytes, int numSplColumns) throws SQLException { |
| long columnValueSize = 0; |
| Map<PColumn,byte[]> columnValues = Maps.newHashMapWithExpectedSize(columnIndexes.length); |
| byte[][] pkValues = new byte[table.getPKColumns().size()][]; |
| // If the table uses salting, the first byte is the salting byte, set to an empty array |
| // here and we will fill in the byte later in PRowImpl. |
| if (table.getBucketNum() != null) { |
| pkValues[0] = new byte[] {0}; |
| } |
| for(int i = 0; i < numSplColumns; i++) { |
| pkValues[i + (table.getBucketNum() != null ? 1 : 0)] = values[i]; |
| } |
| Long rowTimestamp = null; // case when the table doesn't have a row timestamp column |
| RowTimestampColInfo rowTsColInfo = new RowTimestampColInfo(useServerTimestamp, rowTimestamp); |
| for (int i = 0, j = numSplColumns; j < values.length; j++, i++) { |
| byte[] value = values[j]; |
| PColumn column = table.getColumns().get(columnIndexes[i]); |
| if (SchemaUtil.isPKColumn(column)) { |
| pkValues[pkSlotIndex[i]] = value; |
| if (SchemaUtil.getPKPosition(table, column) == table.getRowTimestampColPos()) { |
| if (!useServerTimestamp) { |
| PColumn rowTimestampCol = table.getPKColumns().get(table.getRowTimestampColPos()); |
| rowTimestamp = PLong.INSTANCE.getCodec().decodeLong(value, 0, rowTimestampCol.getSortOrder()); |
| if (rowTimestamp < 0) { |
| throw new IllegalDataException("Value of a column designated as ROW_TIMESTAMP cannot be less than zero"); |
| } |
| rowTsColInfo = new RowTimestampColInfo(useServerTimestamp, rowTimestamp); |
| } |
| } |
| } else { |
| columnValues.put(column, value); |
| columnValueSize += (column.getEstimatedSize() + value.length); |
| } |
| } |
| ImmutableBytesPtr ptr = new ImmutableBytesPtr(); |
| table.newKey(ptr, pkValues); |
| if (table.getIndexType() == IndexType.LOCAL && maintainer != null) { |
| byte[] rowKey = maintainer.buildDataRowKey(ptr, viewConstants); |
| HRegionLocation region = |
| statement.getConnection().getQueryServices() |
| .getTableRegionLocation(table.getParentName().getBytes(), rowKey); |
| byte[] regionPrefix = |
| region.getRegion().getStartKey().length == 0 ? new byte[region |
| .getRegion().getEndKey().length] : region.getRegion() |
| .getStartKey(); |
| if (regionPrefix.length != 0) { |
| ptr.set(ScanRanges.prefixKey(ptr.get(), 0, ptr.getLength(), regionPrefix, |
| regionPrefix.length)); |
| } |
| } |
| mutation.put(ptr, new RowMutationState(columnValues, columnValueSize, statement.getConnection().getStatementExecutionCounter(), rowTsColInfo, onDupKeyBytes)); |
| } |
| |
| public static MutationState upsertSelect(StatementContext childContext, TableRef tableRef, RowProjector projector, |
| ResultIterator iterator, int[] columnIndexes, int[] pkSlotIndexes, boolean useServerTimestamp, boolean prefixSysColValues) throws SQLException { |
| PhoenixStatement statement = childContext.getStatement(); |
| PhoenixConnection connection = statement.getConnection(); |
| ConnectionQueryServices services = connection.getQueryServices(); |
| int maxSize = services.getProps().getInt(QueryServices.MAX_MUTATION_SIZE_ATTRIB, |
| QueryServicesOptions.DEFAULT_MAX_MUTATION_SIZE); |
| int maxSizeBytes = |
| services.getProps().getInt(QueryServices.MAX_MUTATION_SIZE_BYTES_ATTRIB, |
| QueryServicesOptions.DEFAULT_MAX_MUTATION_SIZE_BYTES); |
| int batchSize = Math.min(connection.getMutateBatchSize(), maxSize); |
| // we automatically flush the mutations when either auto commit is enabled, or |
| // the target table is transactional (in that case changes are not visible until we commit) |
| final boolean autoFlush = connection.getAutoCommit() || tableRef.getTable().isTransactional(); |
| int sizeOffset = 0; |
| int numSplColumns = |
| (tableRef.getTable().isMultiTenant() ? 1 : 0) |
| + (tableRef.getTable().getViewIndexId() != null ? 1 : 0); |
| byte[][] values = new byte[columnIndexes.length + numSplColumns][]; |
| if(prefixSysColValues) { |
| int i = 0; |
| if(tableRef.getTable().isMultiTenant()) { |
| values[i++] = connection.getTenantId().getBytes(); |
| } |
| if(tableRef.getTable().getViewIndexId() != null) { |
| values[i++] = PSmallint.INSTANCE.toBytes(tableRef.getTable().getViewIndexId()); |
| } |
| } |
| int rowCount = 0; |
| MultiRowMutationState mutation = new MultiRowMutationState(batchSize); |
| PTable table = tableRef.getTable(); |
| IndexMaintainer indexMaintainer = null; |
| byte[][] viewConstants = null; |
| if (table.getIndexType() == IndexType.LOCAL) { |
| PTable parentTable = |
| statement |
| .getConnection() |
| .getMetaDataCache() |
| .getTableRef( |
| new PTableKey(statement.getConnection().getTenantId(), table |
| .getParentName().getString())).getTable(); |
| indexMaintainer = table.getIndexMaintainer(parentTable, connection); |
| viewConstants = IndexUtil.getViewConstants(parentTable); |
| } |
| try (ResultSet rs = new PhoenixResultSet(iterator, projector, childContext)) { |
| ImmutableBytesWritable ptr = new ImmutableBytesWritable(); |
| while (rs.next()) { |
| for (int i = 0, j = numSplColumns; j < values.length; j++, i++) { |
| PColumn column = table.getColumns().get(columnIndexes[i]); |
| byte[] bytes = rs.getBytes(i + 1); |
| ptr.set(bytes == null ? ByteUtil.EMPTY_BYTE_ARRAY : bytes); |
| Object value = rs.getObject(i + 1); |
| int rsPrecision = rs.getMetaData().getPrecision(i + 1); |
| Integer precision = rsPrecision == 0 ? null : rsPrecision; |
| int rsScale = rs.getMetaData().getScale(i + 1); |
| Integer scale = rsScale == 0 ? null : rsScale; |
| // We are guaranteed that the two column will have compatible types, |
| // as we checked that before. |
| if (!column.getDataType().isSizeCompatible(ptr, value, column.getDataType(), SortOrder.getDefault(), precision, |
| scale, column.getMaxLength(), column.getScale())) { throw new SQLExceptionInfo.Builder( |
| SQLExceptionCode.DATA_EXCEEDS_MAX_CAPACITY).setColumnName(column.getName().getString()) |
| .setMessage("value=" + column.getDataType().toStringLiteral(ptr, null)).build() |
| .buildException(); } |
| column.getDataType().coerceBytes(ptr, value, column.getDataType(), |
| precision, scale, SortOrder.getDefault(), |
| column.getMaxLength(), column.getScale(), column.getSortOrder(), |
| table.rowKeyOrderOptimizable()); |
| values[j] = ByteUtil.copyKeyBytesIfNecessary(ptr); |
| } |
| setValues(values, pkSlotIndexes, columnIndexes, table, mutation, statement, useServerTimestamp, indexMaintainer, viewConstants, null, numSplColumns); |
| rowCount++; |
| // Commit a batch if auto commit is true and we're at our batch size |
| if (autoFlush && rowCount % batchSize == 0) { |
| MutationState state = new MutationState(tableRef, mutation, 0, maxSize, maxSizeBytes, connection); |
| connection.getMutationState().join(state); |
| connection.getMutationState().send(); |
| mutation.clear(); |
| } |
| } |
| |
| if (autoFlush) { |
| // If auto commit is true, this last batch will be committed upon return |
| sizeOffset = rowCount / batchSize * batchSize; |
| } |
| return new MutationState(tableRef, mutation, sizeOffset, maxSize, |
| maxSizeBytes, connection); |
| } |
| } |
| |
| private static class UpsertingParallelIteratorFactory extends MutatingParallelIteratorFactory { |
| private RowProjector projector; |
| private int[] columnIndexes; |
| private int[] pkSlotIndexes; |
| private final TableRef tableRef; |
| private final boolean useSeverTimestamp; |
| |
| private UpsertingParallelIteratorFactory (PhoenixConnection connection, TableRef tableRef, boolean useServerTimestamp) { |
| super(connection); |
| this.tableRef = tableRef; |
| this.useSeverTimestamp = useServerTimestamp; |
| } |
| |
| @Override |
| protected MutationState mutate(StatementContext parentContext, ResultIterator iterator, PhoenixConnection connection) throws SQLException { |
| if (parentContext.getSequenceManager().getSequenceCount() > 0) { |
| throw new IllegalStateException("Cannot pipeline upsert when sequence is referenced"); |
| } |
| PhoenixStatement statement = new PhoenixStatement(connection); |
| /* |
| * We don't want to collect any read metrics within the child context. This is because any read metrics that |
| * need to be captured are already getting collected in the parent statement context enclosed in the result |
| * iterator being used for reading rows out. |
| */ |
| StatementContext childContext = new StatementContext(statement, false); |
| // Clone the row projector as it's not thread safe and would be used simultaneously by |
| // multiple threads otherwise. |
| MutationState state = upsertSelect(childContext, tableRef, projector.cloneIfNecessary(), iterator, columnIndexes, pkSlotIndexes, useSeverTimestamp, false); |
| return state; |
| } |
| |
| public void setRowProjector(RowProjector projector) { |
| this.projector = projector; |
| } |
| public void setColumnIndexes(int[] columnIndexes) { |
| this.columnIndexes = columnIndexes; |
| } |
| public void setPkSlotIndexes(int[] pkSlotIndexes) { |
| this.pkSlotIndexes = pkSlotIndexes; |
| } |
| } |
| |
| private final PhoenixStatement statement; |
| private final Operation operation; |
| |
| public UpsertCompiler(PhoenixStatement statement, Operation operation) { |
| this.statement = statement; |
| this.operation = operation; |
| } |
| |
| private static LiteralParseNode getNodeForRowTimestampColumn(PColumn col) { |
| PDataType type = col.getDataType(); |
| long dummyValue = 0L; |
| if (type.isCoercibleTo(PTimestamp.INSTANCE)) { |
| return new LiteralParseNode(new Timestamp(dummyValue), PTimestamp.INSTANCE); |
| } else if (type == PLong.INSTANCE || type == PUnsignedLong.INSTANCE) { |
| return new LiteralParseNode(dummyValue, PLong.INSTANCE); |
| } |
| throw new IllegalArgumentException(); |
| } |
| |
| public MutationPlan compile(UpsertStatement upsert) throws SQLException { |
| final PhoenixConnection connection = statement.getConnection(); |
| ConnectionQueryServices services = connection.getQueryServices(); |
| final int maxSize = services.getProps().getInt(QueryServices.MAX_MUTATION_SIZE_ATTRIB,QueryServicesOptions.DEFAULT_MAX_MUTATION_SIZE); |
| final int maxSizeBytes = services.getProps().getInt(QueryServices.MAX_MUTATION_SIZE_BYTES_ATTRIB,QueryServicesOptions.DEFAULT_MAX_MUTATION_SIZE_BYTES); |
| List<ColumnName> columnNodes = upsert.getColumns(); |
| TableRef tableRefToBe = null; |
| PTable table = null; |
| Set<PColumn> addViewColumnsToBe = Collections.emptySet(); |
| Set<PColumn> overlapViewColumnsToBe = Collections.emptySet(); |
| List<PColumn> allColumnsToBe = Collections.emptyList(); |
| boolean isTenantSpecific = false; |
| boolean isSharedViewIndex = false; |
| String tenantIdStr = null; |
| ColumnResolver resolver = null; |
| int[] columnIndexesToBe; |
| int nColumnsToSet = 0; |
| int[] pkSlotIndexesToBe; |
| List<ParseNode> valueNodes = upsert.getValues(); |
| List<PColumn> targetColumns; |
| NamedTableNode tableNode = upsert.getTable(); |
| String tableName = tableNode.getName().getTableName(); |
| String schemaName = tableNode.getName().getSchemaName(); |
| QueryPlan queryPlanToBe = null; |
| int nValuesToSet; |
| boolean sameTable = false; |
| boolean runOnServer = false; |
| boolean serverUpsertSelectEnabled = |
| services.getProps().getBoolean(QueryServices.ENABLE_SERVER_UPSERT_SELECT, |
| QueryServicesOptions.DEFAULT_ENABLE_SERVER_UPSERT_SELECT); |
| boolean allowServerMutations = |
| services.getProps().getBoolean(QueryServices.ENABLE_SERVER_SIDE_UPSERT_MUTATIONS, |
| QueryServicesOptions.DEFAULT_ENABLE_SERVER_SIDE_UPSERT_MUTATIONS); |
| UpsertingParallelIteratorFactory parallelIteratorFactoryToBe = null; |
| boolean useServerTimestampToBe = false; |
| |
| |
| resolver = FromCompiler.getResolverForMutation(upsert, connection); |
| tableRefToBe = resolver.getTables().get(0); |
| table = tableRefToBe.getTable(); |
| // Cannot update: |
| // - read-only VIEW |
| // - transactional table with a connection having an SCN |
| // - table with indexes and SCN set |
| // - tables with ROW_TIMESTAMP columns |
| if (table.getType() == PTableType.VIEW && table.getViewType().isReadOnly()) { |
| throw new ReadOnlyTableException(schemaName,tableName); |
| } else if (connection.isBuildingIndex() && table.getType() != PTableType.INDEX) { |
| throw new SQLExceptionInfo.Builder(SQLExceptionCode.ONLY_INDEX_UPDATABLE_AT_SCN) |
| .setSchemaName(schemaName) |
| .setTableName(tableName) |
| .build().buildException(); |
| } else if (table.isTransactional() && connection.getSCN() != null) { |
| throw new SQLExceptionInfo.Builder(SQLExceptionCode |
| .CANNOT_SPECIFY_SCN_FOR_TXN_TABLE) |
| .setSchemaName(schemaName) |
| .setTableName(tableName).build().buildException(); |
| } else if (connection.getSCN() != null && !table.getIndexes().isEmpty() |
| && !connection.isRunningUpgrade() && !connection.isBuildingIndex()) { |
| throw new SQLExceptionInfo |
| .Builder(SQLExceptionCode |
| .CANNOT_UPSERT_WITH_SCN_FOR_TABLE_WITH_INDEXES) |
| .setSchemaName(schemaName) |
| .setTableName(tableName).build().buildException(); |
| } else if(connection.getSCN() != null && !connection.isRunningUpgrade() |
| && !connection.isBuildingIndex() && table.getRowTimestampColPos() >= 0) { |
| throw new SQLExceptionInfo |
| .Builder(SQLExceptionCode |
| .CANNOT_UPSERT_WITH_SCN_FOR_ROW_TIMESTAMP_COLUMN) |
| .setSchemaName(schemaName) |
| .setTableName(tableName).build().buildException(); |
| } |
| boolean isSalted = table.getBucketNum() != null; |
| isTenantSpecific = table.isMultiTenant() && connection.getTenantId() != null; |
| isSharedViewIndex = table.getViewIndexId() != null; |
| tenantIdStr = isTenantSpecific ? connection.getTenantId().getString() : null; |
| int posOffset = isSalted ? 1 : 0; |
| // Setup array of column indexes parallel to values that are going to be set |
| allColumnsToBe = table.getColumns(); |
| |
| nColumnsToSet = 0; |
| if (table.getViewType() == ViewType.UPDATABLE) { |
| addViewColumnsToBe = Sets.newLinkedHashSetWithExpectedSize(allColumnsToBe.size()); |
| for (PColumn column : allColumnsToBe) { |
| if (column.getViewConstant() != null) { |
| addViewColumnsToBe.add(column); |
| } |
| } |
| } |
| ImmutableBytesWritable ptr = new ImmutableBytesWritable(); |
| // Allow full row upsert if no columns or only dynamic ones are specified and values count match |
| if (columnNodes.isEmpty() || columnNodes.size() == upsert.getTable().getDynamicColumns().size()) { |
| nColumnsToSet = allColumnsToBe.size() - posOffset; |
| columnIndexesToBe = new int[nColumnsToSet]; |
| pkSlotIndexesToBe = new int[columnIndexesToBe.length]; |
| targetColumns = Lists.newArrayListWithExpectedSize(columnIndexesToBe.length); |
| targetColumns.addAll(Collections.<PColumn>nCopies(columnIndexesToBe.length, null)); |
| int minPKPos = 0; |
| if (isSharedViewIndex) { |
| PColumn indexIdColumn = table.getPKColumns().get(minPKPos); |
| columnIndexesToBe[minPKPos] = indexIdColumn.getPosition(); |
| targetColumns.set(minPKPos, indexIdColumn); |
| minPKPos++; |
| } |
| if (isTenantSpecific) { |
| PColumn tenantColumn = table.getPKColumns().get(minPKPos); |
| columnIndexesToBe[minPKPos] = tenantColumn.getPosition(); |
| targetColumns.set(minPKPos, tenantColumn); |
| minPKPos++; |
| } |
| for (int i = posOffset, j = 0; i < allColumnsToBe.size(); i++) { |
| PColumn column = allColumnsToBe.get(i); |
| if (SchemaUtil.isPKColumn(column)) { |
| pkSlotIndexesToBe[i-posOffset] = j + posOffset; |
| if (j++ < minPKPos) { // Skip, as it's already been set above |
| continue; |
| } |
| minPKPos = 0; |
| } |
| columnIndexesToBe[i-posOffset+minPKPos] = i; |
| targetColumns.set(i-posOffset+minPKPos, column); |
| } |
| if (!addViewColumnsToBe.isEmpty()) { |
| // All view columns overlap in this case |
| overlapViewColumnsToBe = addViewColumnsToBe; |
| addViewColumnsToBe = Collections.emptySet(); |
| } |
| } else { |
| // Size for worse case |
| int numColsInUpsert = columnNodes.size(); |
| nColumnsToSet = numColsInUpsert + addViewColumnsToBe.size() + (isTenantSpecific ? 1 : 0) + + (isSharedViewIndex ? 1 : 0); |
| columnIndexesToBe = new int[nColumnsToSet]; |
| pkSlotIndexesToBe = new int[columnIndexesToBe.length]; |
| targetColumns = Lists.newArrayListWithExpectedSize(columnIndexesToBe.length); |
| targetColumns.addAll(Collections.<PColumn>nCopies(columnIndexesToBe.length, null)); |
| Arrays.fill(columnIndexesToBe, -1); // TODO: necessary? So we'll get an AIOB exception if it's not replaced |
| Arrays.fill(pkSlotIndexesToBe, -1); // TODO: necessary? So we'll get an AIOB exception if it's not replaced |
| BitSet columnsBeingSet = new BitSet(table.getColumns().size()); |
| int i = 0; |
| if (isSharedViewIndex) { |
| PColumn indexIdColumn = table.getPKColumns().get(i + posOffset); |
| columnsBeingSet.set(columnIndexesToBe[i] = indexIdColumn.getPosition()); |
| pkSlotIndexesToBe[i] = i + posOffset; |
| targetColumns.set(i, indexIdColumn); |
| i++; |
| } |
| // Add tenant column directly, as we don't want to resolve it as this will fail |
| if (isTenantSpecific) { |
| PColumn tenantColumn = table.getPKColumns().get(i + posOffset); |
| columnsBeingSet.set(columnIndexesToBe[i] = tenantColumn.getPosition()); |
| pkSlotIndexesToBe[i] = i + posOffset; |
| targetColumns.set(i, tenantColumn); |
| i++; |
| } |
| for (ColumnName colName : columnNodes) { |
| ColumnRef ref = resolver.resolveColumn(null, colName.getFamilyName(), colName.getColumnName()); |
| PColumn column = ref.getColumn(); |
| if (IndexUtil.getViewConstantValue(column, ptr)) { |
| if (overlapViewColumnsToBe.isEmpty()) { |
| overlapViewColumnsToBe = Sets.newHashSetWithExpectedSize(addViewColumnsToBe.size()); |
| } |
| nColumnsToSet--; |
| overlapViewColumnsToBe.add(column); |
| addViewColumnsToBe.remove(column); |
| } |
| columnsBeingSet.set(columnIndexesToBe[i] = ref.getColumnPosition()); |
| targetColumns.set(i, column); |
| if (SchemaUtil.isPKColumn(column)) { |
| pkSlotIndexesToBe[i] = ref.getPKSlotPosition(); |
| } |
| i++; |
| } |
| for (PColumn column : addViewColumnsToBe) { |
| columnsBeingSet.set(columnIndexesToBe[i] = column.getPosition()); |
| targetColumns.set(i, column); |
| if (SchemaUtil.isPKColumn(column)) { |
| pkSlotIndexesToBe[i] = SchemaUtil.getPKPosition(table, column); |
| } |
| i++; |
| } |
| // If a table has rowtimestamp col, then we always set it. |
| useServerTimestampToBe = table.getRowTimestampColPos() != -1 && !isRowTimestampSet(pkSlotIndexesToBe, table); |
| if (useServerTimestampToBe) { |
| PColumn rowTimestampCol = table.getPKColumns().get(table.getRowTimestampColPos()); |
| // Need to resize columnIndexesToBe and pkSlotIndexesToBe to include this extra column. |
| columnIndexesToBe = Arrays.copyOf(columnIndexesToBe, columnIndexesToBe.length + 1); |
| pkSlotIndexesToBe = Arrays.copyOf(pkSlotIndexesToBe, pkSlotIndexesToBe.length + 1); |
| columnsBeingSet.set(columnIndexesToBe[i] = rowTimestampCol.getPosition()); |
| pkSlotIndexesToBe[i] = table.getRowTimestampColPos(); |
| targetColumns.add(rowTimestampCol); |
| if (valueNodes != null && !valueNodes.isEmpty()) { |
| valueNodes.add(getNodeForRowTimestampColumn(rowTimestampCol)); |
| } |
| nColumnsToSet++; |
| } |
| for (i = posOffset; i < table.getColumns().size(); i++) { |
| PColumn column = table.getColumns().get(i); |
| if (!columnsBeingSet.get(i) && !column.isNullable() && column.getExpressionStr() == null) { |
| throw new ConstraintViolationException(SchemaUtil.getColumnDisplayName(column) + " may not be null"); |
| } |
| } |
| } |
| boolean isAutoCommit = connection.getAutoCommit(); |
| if (valueNodes == null) { |
| SelectStatement select = upsert.getSelect(); |
| assert(select != null); |
| select = SubselectRewriter.flatten(select, connection); |
| ColumnResolver selectResolver = FromCompiler.getResolverForQuery(select, connection, false, upsert.getTable().getName()); |
| select = StatementNormalizer.normalize(select, selectResolver); |
| select = prependTenantAndViewConstants(table, select, tenantIdStr, addViewColumnsToBe, useServerTimestampToBe); |
| SelectStatement transformedSelect = SubqueryRewriter.transform(select, selectResolver, connection); |
| if (transformedSelect != select) { |
| selectResolver = FromCompiler.getResolverForQuery(transformedSelect, connection, false, upsert.getTable().getName()); |
| select = StatementNormalizer.normalize(transformedSelect, selectResolver); |
| } |
| sameTable = !select.isJoin() |
| && tableRefToBe.equals(selectResolver.getTables().get(0)); |
| /* We can run the upsert in a coprocessor if: |
| * 1) from has only 1 table or server UPSERT SELECT is enabled |
| * 2) the select query isn't doing aggregation (which requires a client-side final merge) |
| * 3) autoCommit is on |
| * 4) the table is not immutable with indexes, as the client is the one that figures out the additional |
| * puts for index tables. |
| * 5) no limit clause, as the limit clause requires client-side post processing |
| * 6) no sequences, as sequences imply that the order of upsert must match the order of |
| * selection. TODO: change this and only force client side if there's a ORDER BY on the sequence value |
| * Otherwise, run the query to pull the data from the server |
| * and populate the MutationState (upto a limit). |
| */ |
| if (! (select.isAggregate() || select.isDistinct() || select.getLimit() != null || select.hasSequence()) ) { |
| // We can pipeline the upsert select instead of spooling everything to disk first, |
| // if we don't have any post processing that's required. |
| parallelIteratorFactoryToBe = new UpsertingParallelIteratorFactory(connection, tableRefToBe, useServerTimestampToBe); |
| // If we're in the else, then it's not an aggregate, distinct, limited, or sequence using query, |
| // so we might be able to run it entirely on the server side. |
| // region space managed by region servers. So we bail out on executing on server side. |
| // Disable running upsert select on server side if a table has global mutable secondary indexes on it |
| boolean hasGlobalMutableIndexes = SchemaUtil.hasGlobalIndex(table) && !table.isImmutableRows(); |
| boolean hasWhereSubquery = select.getWhere() != null && select.getWhere().hasSubquery(); |
| runOnServer = (sameTable || (serverUpsertSelectEnabled && !hasGlobalMutableIndexes)) && isAutoCommit |
| // We can run the upsert select for initial index population on server side for transactional |
| // tables since the writes do not need to be done transactionally, since we gate the index |
| // usage on successfully writing all data rows. |
| && (!table.isTransactional() || table.getType() == PTableType.INDEX) |
| && !(table.isImmutableRows() && !table.getIndexes().isEmpty()) |
| && !select.isJoin() && !hasWhereSubquery && table.getRowTimestampColPos() == -1; |
| } |
| runOnServer &= allowServerMutations; |
| // If we may be able to run on the server, add a hint that favors using the data table |
| // if all else is equal. |
| // TODO: it'd be nice if we could figure out in advance if the PK is potentially changing, |
| // as this would disallow running on the server. We currently use the row projector we |
| // get back to figure this out. |
| HintNode hint = upsert.getHint(); |
| if (!upsert.getHint().hasHint(Hint.USE_INDEX_OVER_DATA_TABLE)) { |
| hint = HintNode.create(hint, Hint.USE_DATA_OVER_INDEX_TABLE); |
| } |
| select = SelectStatement.create(select, hint); |
| // Pass scan through if same table in upsert and select so that projection is computed correctly |
| // Use optimizer to choose the best plan |
| QueryCompiler compiler = new QueryCompiler(statement, select, selectResolver, targetColumns, parallelIteratorFactoryToBe, new SequenceManager(statement), true, false, null); |
| queryPlanToBe = compiler.compile(); |
| |
| if (sameTable) { |
| // in the UPSERT INTO X ... SELECT FROM X case enforce the source tableRef's TS |
| // as max TS, so that the query can safely restarted and still work of a snapshot |
| // (so it won't see its own data in case of concurrent splits) |
| // see PHOENIX-4849 |
| long serverTime = selectResolver.getTables().get(0).getCurrentTime(); |
| if (serverTime == QueryConstants.UNSET_TIMESTAMP) { |
| // if this is the first time this table is resolved the ref's current time might not be defined, yet |
| // in that case force an RPC to get the server time |
| serverTime = new MetaDataClient(connection).getCurrentTime(schemaName, tableName); |
| } |
| Scan scan = queryPlanToBe.getContext().getScan(); |
| ScanUtil.setTimeRange(scan, scan.getTimeRange().getMin(), serverTime); |
| } |
| // This is post-fix: if the tableRef is a projected table, this means there are post-processing |
| // steps and parallelIteratorFactory did not take effect. |
| if (queryPlanToBe.getTableRef().getTable().getType() == PTableType.PROJECTED || queryPlanToBe.getTableRef().getTable().getType() == PTableType.SUBQUERY) { |
| parallelIteratorFactoryToBe = null; |
| } |
| nValuesToSet = queryPlanToBe.getProjector().getColumnCount(); |
| // Cannot auto commit if doing aggregation or topN or salted |
| // Salted causes problems because the row may end up living on a different region |
| } else { |
| nValuesToSet = valueNodes.size() + addViewColumnsToBe.size() + (isTenantSpecific ? 1 : 0) + (isSharedViewIndex ? 1 : 0); |
| } |
| // Resize down to allow a subset of columns to be specifiable |
| if (columnNodes.isEmpty() && columnIndexesToBe.length >= nValuesToSet) { |
| nColumnsToSet = nValuesToSet; |
| columnIndexesToBe = Arrays.copyOf(columnIndexesToBe, nValuesToSet); |
| pkSlotIndexesToBe = Arrays.copyOf(pkSlotIndexesToBe, nValuesToSet); |
| for (int i = posOffset + nValuesToSet; i < table.getColumns().size(); i++) { |
| PColumn column = table.getColumns().get(i); |
| if (!column.isNullable() && column.getExpressionStr() == null) { |
| throw new ConstraintViolationException(SchemaUtil.getColumnDisplayName(column) + " may not be null"); |
| } |
| } |
| } |
| |
| if (nValuesToSet != nColumnsToSet) { |
| // We might have added columns, so refresh cache and try again if stale. |
| // We have logic to catch MetaNotFoundException and refresh cache in PhoenixStatement |
| // Note that this check is not really sufficient, as a column could have |
| // been removed and the added back and we wouldn't detect that here. |
| throw new UpsertColumnsValuesMismatchException(schemaName, tableName, |
| "Numbers of columns: " + nColumnsToSet + ". Number of values: " + nValuesToSet); |
| } |
| final QueryPlan originalQueryPlan = queryPlanToBe; |
| RowProjector projectorToBe = null; |
| // Optimize only after all checks have been performed |
| if (valueNodes == null) { |
| queryPlanToBe = new QueryOptimizer(services).optimize(queryPlanToBe, statement, targetColumns, parallelIteratorFactoryToBe); |
| projectorToBe = queryPlanToBe.getProjector(); |
| } |
| final List<PColumn> allColumns = allColumnsToBe; |
| final RowProjector projector = projectorToBe; |
| final QueryPlan queryPlan = queryPlanToBe; |
| final TableRef tableRef = tableRefToBe; |
| final Set<PColumn> addViewColumns = addViewColumnsToBe; |
| final Set<PColumn> overlapViewColumns = overlapViewColumnsToBe; |
| final UpsertingParallelIteratorFactory parallelIteratorFactory = parallelIteratorFactoryToBe; |
| final int[] columnIndexes = columnIndexesToBe; |
| final int[] pkSlotIndexes = pkSlotIndexesToBe; |
| final boolean useServerTimestamp = useServerTimestampToBe; |
| if (table.getRowTimestampColPos() == -1 && useServerTimestamp) { |
| throw new IllegalStateException("For a table without row timestamp column, useServerTimestamp cannot be true"); |
| } |
| // TODO: break this up into multiple functions |
| //////////////////////////////////////////////////////////////////// |
| // UPSERT SELECT |
| ///////////////////////////////////////////////////////////////////// |
| if (valueNodes == null) { |
| // Before we re-order, check that for updatable view columns |
| // the projected expression either matches the column name or |
| // is a constant with the same required value. |
| throwIfNotUpdatable(tableRef, overlapViewColumnsToBe, targetColumns, projector, sameTable); |
| |
| //////////////////////////////////////////////////////////////////// |
| // UPSERT SELECT run server-side (maybe) |
| ///////////////////////////////////////////////////////////////////// |
| if (runOnServer) { |
| // At most this array will grow bigger by the number of PK columns |
| int[] allColumnsIndexes = Arrays.copyOf(columnIndexes, columnIndexes.length + nValuesToSet); |
| int[] reverseColumnIndexes = new int[table.getColumns().size()]; |
| List<Expression> projectedExpressions = Lists.newArrayListWithExpectedSize(reverseColumnIndexes.length); |
| Arrays.fill(reverseColumnIndexes, -1); |
| for (int i =0; i < nValuesToSet; i++) { |
| projectedExpressions.add(projector.getColumnProjector(i).getExpression()); |
| reverseColumnIndexes[columnIndexes[i]] = i; |
| } |
| /* |
| * Order projected columns and projected expressions with PK columns |
| * leading order by slot position |
| */ |
| int offset = table.getBucketNum() == null ? 0 : 1; |
| for (int i = 0; i < table.getPKColumns().size() - offset; i++) { |
| PColumn column = table.getPKColumns().get(i + offset); |
| int pos = reverseColumnIndexes[column.getPosition()]; |
| if (pos == -1) { |
| // Last PK column may be fixed width and nullable |
| // We don't want to insert a null expression b/c |
| // it's not valid to set a fixed width type to null. |
| if (column.getDataType().isFixedWidth()) { |
| continue; |
| } |
| // Add literal null for missing PK columns |
| pos = projectedExpressions.size(); |
| Expression literalNull = LiteralExpression.newConstant(null, column.getDataType(), Determinism.ALWAYS); |
| projectedExpressions.add(literalNull); |
| allColumnsIndexes[pos] = column.getPosition(); |
| } |
| // Swap select expression at pos with i |
| Collections.swap(projectedExpressions, i, pos); |
| // Swap column indexes and reverse column indexes too |
| int tempPos = allColumnsIndexes[i]; |
| allColumnsIndexes[i] = allColumnsIndexes[pos]; |
| allColumnsIndexes[pos] = tempPos; |
| reverseColumnIndexes[tempPos] = pos; |
| reverseColumnIndexes[i] = i; |
| } |
| // If any pk slots are changing and server side UPSERT SELECT is disabled, do not run on server |
| if (!serverUpsertSelectEnabled && ExpressionUtil |
| .isPkPositionChanging(new TableRef(table), projectedExpressions)) { |
| runOnServer = false; |
| } |
| //////////////////////////////////////////////////////////////////// |
| // UPSERT SELECT run server-side |
| ///////////////////////////////////////////////////////////////////// |
| if (runOnServer) { |
| // Iterate through columns being projected |
| List<PColumn> projectedColumns = Lists.newArrayListWithExpectedSize(projectedExpressions.size()); |
| int posOff = table.getBucketNum() != null ? 1 : 0; |
| for (int i = 0 ; i < projectedExpressions.size(); i++) { |
| // Must make new column if position has changed |
| PColumn column = allColumns.get(allColumnsIndexes[i]); |
| projectedColumns.add(column.getPosition() == i + posOff ? column : new PColumnImpl(column, i + posOff)); |
| } |
| // Build table from projectedColumns |
| // Hack to add default column family to be used on server in case no value column is projected. |
| PTable projectedTable = PTableImpl.builderWithColumns(table, projectedColumns) |
| .setExcludedColumns(ImmutableList.of()) |
| .setDefaultFamilyName(PNameFactory.newName(SchemaUtil.getEmptyColumnFamily(table))) |
| .build(); |
| |
| SelectStatement select = SelectStatement.create(SelectStatement.COUNT_ONE, upsert.getHint()); |
| StatementContext statementContext = queryPlan.getContext(); |
| RowProjector aggProjectorToBe = ProjectionCompiler.compile(statementContext, select, GroupBy |
| .EMPTY_GROUP_BY); |
| statementContext.getAggregationManager().compile(queryPlan.getContext() |
| ,GroupBy.EMPTY_GROUP_BY); |
| if (queryPlan.getProjector().projectEveryRow()) { |
| aggProjectorToBe = new RowProjector(aggProjectorToBe,true); |
| } |
| final RowProjector aggProjector = aggProjectorToBe; |
| |
| /* |
| * Transfer over PTable representing subset of columns selected, but all PK columns. |
| * Move columns setting PK first in pkSlot order, adding LiteralExpression of null for any missing ones. |
| * Transfer over List<Expression> for projection. |
| * In region scan, evaluate expressions in order, collecting first n columns for PK and collection non PK in mutation Map |
| * Create the PRow and get the mutations, adding them to the batch |
| */ |
| final StatementContext context = queryPlan.getContext(); |
| final Scan scan = context.getScan(); |
| scan.setAttribute(BaseScannerRegionObserver.UPSERT_SELECT_TABLE, UngroupedAggregateRegionObserver.serialize(projectedTable)); |
| scan.setAttribute(BaseScannerRegionObserver.UPSERT_SELECT_EXPRS, UngroupedAggregateRegionObserver.serialize(projectedExpressions)); |
| |
| // Ignore order by - it has no impact |
| final QueryPlan aggPlan = new AggregatePlan(context, select, statementContext.getCurrentTable(), aggProjector, null,null, OrderBy.EMPTY_ORDER_BY, null, GroupBy.EMPTY_GROUP_BY, null, originalQueryPlan); |
| return new ServerUpsertSelectMutationPlan(queryPlan, tableRef, originalQueryPlan, context, connection, scan, aggPlan, aggProjector, maxSize, maxSizeBytes); |
| } |
| } |
| //////////////////////////////////////////////////////////////////// |
| // UPSERT SELECT run client-side |
| ///////////////////////////////////////////////////////////////////// |
| return new ClientUpsertSelectMutationPlan(queryPlan, tableRef, originalQueryPlan, parallelIteratorFactory, projector, columnIndexes, pkSlotIndexes, useServerTimestamp, maxSize, maxSizeBytes); |
| } |
| |
| |
| //////////////////////////////////////////////////////////////////// |
| // UPSERT VALUES |
| ///////////////////////////////////////////////////////////////////// |
| final byte[][] values = new byte[nValuesToSet][]; |
| int nodeIndex = 0; |
| if (isSharedViewIndex) { |
| values[nodeIndex++] = table.getviewIndexIdType().toBytes(table.getViewIndexId()); |
| } |
| if (isTenantSpecific) { |
| PName tenantId = connection.getTenantId(); |
| values[nodeIndex++] = ScanUtil.getTenantIdBytes(table.getRowKeySchema(), table.getBucketNum() != null, tenantId, isSharedViewIndex); |
| } |
| |
| final int nodeIndexOffset = nodeIndex; |
| // Allocate array based on size of all columns in table, |
| // since some values may not be set (if they're nullable). |
| final StatementContext context = new StatementContext(statement, resolver, new Scan(), new SequenceManager(statement)); |
| UpsertValuesCompiler expressionBuilder = new UpsertValuesCompiler(context); |
| final List<Expression> constantExpressions = Lists.newArrayListWithExpectedSize(valueNodes.size()); |
| // First build all the expressions, as with sequences we want to collect them all first |
| // and initialize them in one batch |
| for (ParseNode valueNode : valueNodes) { |
| if (!valueNode.isStateless()) { |
| throw new SQLExceptionInfo.Builder(SQLExceptionCode.VALUE_IN_UPSERT_NOT_CONSTANT).build().buildException(); |
| } |
| PColumn column = allColumns.get(columnIndexes[nodeIndex]); |
| expressionBuilder.setColumn(column); |
| Expression expression = valueNode.accept(expressionBuilder); |
| if (expression.getDataType() != null && !expression.getDataType().isCastableTo(column.getDataType())) { |
| throw TypeMismatchException.newException( |
| expression.getDataType(), column.getDataType(), "expression: " |
| + expression.toString() + " in column " + column); |
| } |
| constantExpressions.add(expression); |
| nodeIndex++; |
| } |
| byte[] onDupKeyBytesToBe = null; |
| List<Pair<ColumnName,ParseNode>> onDupKeyPairs = upsert.getOnDupKeyPairs(); |
| if (onDupKeyPairs != null) { |
| if (table.isImmutableRows()) { |
| throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_USE_ON_DUP_KEY_FOR_IMMUTABLE) |
| .setSchemaName(table.getSchemaName().getString()) |
| .setTableName(table.getTableName().getString()) |
| .build().buildException(); |
| } |
| if (table.isTransactional()) { |
| throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_USE_ON_DUP_KEY_FOR_TRANSACTIONAL) |
| .setSchemaName(table.getSchemaName().getString()) |
| .setTableName(table.getTableName().getString()) |
| .build().buildException(); |
| } |
| if (connection.getSCN() != null) { |
| throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_SET_SCN_IN_ON_DUP_KEY) |
| .setSchemaName(table.getSchemaName().getString()) |
| .setTableName(table.getTableName().getString()) |
| .build().buildException(); |
| } |
| if (SchemaUtil.hasGlobalIndex(table)) { |
| throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_USE_ON_DUP_KEY_WITH_GLOBAL_IDX) |
| .setSchemaName(table.getSchemaName().getString()) |
| .setTableName(table.getTableName().getString()) |
| .build().buildException(); |
| } |
| if (onDupKeyPairs.isEmpty()) { // ON DUPLICATE KEY IGNORE |
| onDupKeyBytesToBe = PhoenixIndexBuilder.serializeOnDupKeyIgnore(); |
| } else { // ON DUPLICATE KEY UPDATE; |
| int position = table.getBucketNum() == null ? 0 : 1; |
| UpdateColumnCompiler compiler = new UpdateColumnCompiler(context); |
| int nColumns = onDupKeyPairs.size(); |
| List<Expression> updateExpressions = Lists.newArrayListWithExpectedSize(nColumns); |
| LinkedHashSet<PColumn> updateColumns = Sets.newLinkedHashSetWithExpectedSize(nColumns + 1); |
| updateColumns.add(new PColumnImpl( |
| table.getPKColumns().get(position).getName(), // Use first PK column name as we know it won't conflict with others |
| null, PVarbinary.INSTANCE, null, null, false, position, SortOrder.getDefault(), 0, null, false, null, false, false, null, table.getPKColumns().get(position).getTimestamp())); |
| position++; |
| for (Pair<ColumnName,ParseNode> columnPair : onDupKeyPairs) { |
| ColumnName colName = columnPair.getFirst(); |
| PColumn updateColumn = resolver.resolveColumn(null, colName.getFamilyName(), colName.getColumnName()).getColumn(); |
| if (SchemaUtil.isPKColumn(updateColumn)) { |
| throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_UPDATE_PK_ON_DUP_KEY) |
| .setSchemaName(table.getSchemaName().getString()) |
| .setTableName(table.getTableName().getString()) |
| .setColumnName(updateColumn.getName().getString()) |
| .build().buildException(); |
| } |
| final int columnPosition = position++; |
| if (!updateColumns.add(new DelegateColumn(updateColumn) { |
| @Override |
| public int getPosition() { |
| return columnPosition; |
| } |
| })) { |
| throw new SQLExceptionInfo.Builder(SQLExceptionCode.DUPLICATE_COLUMN_IN_ON_DUP_KEY) |
| .setSchemaName(table.getSchemaName().getString()) |
| .setTableName(table.getTableName().getString()) |
| .setColumnName(updateColumn.getName().getString()) |
| .build().buildException(); |
| }; |
| ParseNode updateNode = columnPair.getSecond(); |
| compiler.setColumn(updateColumn); |
| Expression updateExpression = updateNode.accept(compiler); |
| // Check that updateExpression is coercible to updateColumn |
| if (updateExpression.getDataType() != null && !updateExpression.getDataType().isCastableTo(updateColumn.getDataType())) { |
| throw TypeMismatchException.newException( |
| updateExpression.getDataType(), updateColumn.getDataType(), "expression: " |
| + updateExpression.toString() + " for column " + updateColumn); |
| } |
| if (compiler.isAggregate()) { |
| throw new SQLExceptionInfo.Builder(SQLExceptionCode.AGGREGATION_NOT_ALLOWED_IN_ON_DUP_KEY) |
| .setSchemaName(table.getSchemaName().getString()) |
| .setTableName(table.getTableName().getString()) |
| .setColumnName(updateColumn.getName().getString()) |
| .build().buildException(); |
| } |
| updateExpressions.add(updateExpression); |
| } |
| PTable onDupKeyTable = PTableImpl.builderWithColumns(table, updateColumns) |
| .build(); |
| onDupKeyBytesToBe = PhoenixIndexBuilder.serializeOnDupKeyUpdate(onDupKeyTable, updateExpressions); |
| } |
| } |
| final byte[] onDupKeyBytes = onDupKeyBytesToBe; |
| |
| return new UpsertValuesMutationPlan(context, tableRef, nodeIndexOffset, constantExpressions, |
| allColumns, columnIndexes, overlapViewColumns, values, addViewColumns, |
| connection, pkSlotIndexes, useServerTimestamp, onDupKeyBytes, maxSize, maxSizeBytes); |
| } |
| |
| private static boolean isRowTimestampSet(int[] pkSlotIndexes, PTable table) { |
| checkArgument(table.getRowTimestampColPos() != -1, "Call this method only for tables with row timestamp column"); |
| int rowTimestampColPKSlot = table.getRowTimestampColPos(); |
| for (int pkSlot : pkSlotIndexes) { |
| if (pkSlot == rowTimestampColPKSlot) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| private static class UpdateColumnCompiler extends ExpressionCompiler { |
| private PColumn column; |
| |
| private UpdateColumnCompiler(StatementContext context) { |
| super(context); |
| } |
| |
| public void setColumn(PColumn column) { |
| this.column = column; |
| } |
| |
| @Override |
| public Expression visit(BindParseNode node) throws SQLException { |
| if (isTopLevel()) { |
| context.getBindManager().addParamMetaData(node, column); |
| Object value = context.getBindManager().getBindValue(node); |
| return LiteralExpression.newConstant(value, column.getDataType(), column.getSortOrder(), Determinism.ALWAYS); |
| } |
| return super.visit(node); |
| } |
| |
| @Override |
| public Expression visit(LiteralParseNode node) throws SQLException { |
| if (isTopLevel()) { |
| return LiteralExpression.newConstant(node.getValue(), column.getDataType(), column.getSortOrder(), Determinism.ALWAYS); |
| } |
| return super.visit(node); |
| } |
| } |
| |
| private static class UpsertValuesCompiler extends UpdateColumnCompiler { |
| private UpsertValuesCompiler(StatementContext context) { |
| super(context); |
| } |
| |
| @Override |
| public Expression visit(SequenceValueParseNode node) throws SQLException { |
| return context.getSequenceManager().newSequenceReference(node); |
| } |
| } |
| |
| |
| private static SelectStatement prependTenantAndViewConstants(PTable table, SelectStatement select, String tenantId, Set<PColumn> addViewColumns, boolean useServerTimestamp) { |
| if ((!table.isMultiTenant() || tenantId == null) && table.getViewIndexId() == null && addViewColumns.isEmpty() && !useServerTimestamp) { |
| return select; |
| } |
| List<AliasedNode> selectNodes = newArrayListWithCapacity(select.getSelect().size() + 1 + addViewColumns.size()); |
| if (table.getViewIndexId() != null) { |
| selectNodes.add(new AliasedNode(null, new LiteralParseNode(table.getViewIndexId()))); |
| } |
| if (table.isMultiTenant() && tenantId != null) { |
| selectNodes.add(new AliasedNode(null, new LiteralParseNode(tenantId))); |
| } |
| selectNodes.addAll(select.getSelect()); |
| for (PColumn column : addViewColumns) { |
| byte[] byteValue = column.getViewConstant(); |
| Object value = column.getDataType().toObject(byteValue, 0, byteValue.length-1); |
| selectNodes.add(new AliasedNode(null, new LiteralParseNode(value))); |
| } |
| if (useServerTimestamp) { |
| PColumn rowTimestampCol = table.getPKColumns().get(table.getRowTimestampColPos()); |
| selectNodes.add(new AliasedNode(null, getNodeForRowTimestampColumn(rowTimestampCol))); |
| } |
| return SelectStatement.create(select, selectNodes); |
| } |
| |
| /** |
| * Check that none of no columns in our updatable VIEW are changing values. |
| * @param tableRef |
| * @param overlapViewColumns |
| * @param targetColumns |
| * @param projector |
| * @throws SQLException |
| */ |
| private static void throwIfNotUpdatable(TableRef tableRef, Set<PColumn> overlapViewColumns, |
| List<PColumn> targetColumns, RowProjector projector, boolean sameTable) throws SQLException { |
| PTable table = tableRef.getTable(); |
| if (table.getViewType() == ViewType.UPDATABLE && !overlapViewColumns.isEmpty()) { |
| ImmutableBytesWritable ptr = new ImmutableBytesWritable(); |
| for (int i = 0; i < targetColumns.size(); i++) { |
| PColumn targetColumn = targetColumns.get(i); |
| if (overlapViewColumns.contains(targetColumn)) { |
| Expression source = projector.getColumnProjector(i).getExpression(); |
| if (source.isStateless()) { |
| source.evaluate(null, ptr); |
| if (Bytes.compareTo(ptr.get(), ptr.getOffset(), ptr.getLength(), targetColumn.getViewConstant(), 0, targetColumn.getViewConstant().length-1) == 0) { |
| continue; |
| } |
| } |
| throw new SQLExceptionInfo.Builder( |
| SQLExceptionCode.CANNOT_UPDATE_VIEW_COLUMN) |
| .setColumnName(targetColumn.getName().getString()) |
| .build().buildException(); |
| } |
| } |
| } |
| } |
| |
| private class ServerUpsertSelectMutationPlan implements MutationPlan { |
| private final QueryPlan queryPlan; |
| private final TableRef tableRef; |
| private final QueryPlan originalQueryPlan; |
| private final StatementContext context; |
| private final PhoenixConnection connection; |
| private final Scan scan; |
| private final QueryPlan aggPlan; |
| private final RowProjector aggProjector; |
| private final int maxSize; |
| private final int maxSizeBytes; |
| |
| public ServerUpsertSelectMutationPlan(QueryPlan queryPlan, TableRef tableRef, QueryPlan originalQueryPlan, |
| StatementContext context, PhoenixConnection connection, |
| Scan scan, QueryPlan aggPlan, RowProjector aggProjector, |
| int maxSize, int maxSizeBytes) { |
| this.queryPlan = queryPlan; |
| this.tableRef = tableRef; |
| this.originalQueryPlan = originalQueryPlan; |
| this.context = context; |
| this.connection = connection; |
| this.scan = scan; |
| this.aggPlan = aggPlan; |
| this.aggProjector = aggProjector; |
| this.maxSize = maxSize; |
| this.maxSizeBytes = maxSizeBytes; |
| } |
| |
| @Override |
| public ParameterMetaData getParameterMetaData() { |
| return queryPlan.getContext().getBindManager().getParameterMetaData(); |
| } |
| |
| @Override |
| public StatementContext getContext() { |
| return queryPlan.getContext(); |
| } |
| |
| @Override |
| public TableRef getTargetRef() { |
| return tableRef; |
| } |
| |
| @Override |
| public QueryPlan getQueryPlan() { |
| return aggPlan; |
| } |
| |
| @Override |
| public Set<TableRef> getSourceRefs() { |
| return originalQueryPlan.getSourceRefs(); |
| } |
| |
| @Override |
| public Operation getOperation() { |
| return operation; |
| } |
| |
| @Override |
| public MutationState execute() throws SQLException { |
| ImmutableBytesWritable ptr = context.getTempPtr(); |
| PTable table = tableRef.getTable(); |
| table.getIndexMaintainers(ptr, context.getConnection()); |
| byte[] txState = table.isTransactional() ? |
| connection.getMutationState().encodeTransaction() : ByteUtil.EMPTY_BYTE_ARRAY; |
| |
| ScanUtil.setClientVersion(scan, MetaDataProtocol.PHOENIX_VERSION); |
| if (aggPlan.getTableRef().getTable().isTransactional() |
| || (table.getType() == PTableType.INDEX && table.isTransactional())) { |
| scan.setAttribute(BaseScannerRegionObserver.TX_STATE, txState); |
| } |
| if (ptr.getLength() > 0) { |
| byte[] uuidValue = ServerCacheClient.generateId(); |
| scan.setAttribute(PhoenixIndexCodec.INDEX_UUID, uuidValue); |
| scan.setAttribute(PhoenixIndexCodec.INDEX_PROTO_MD, ptr.get()); |
| } |
| ResultIterator iterator = aggPlan.iterator(); |
| try { |
| Tuple row = iterator.next(); |
| final long mutationCount = (Long) aggProjector.getColumnProjector(0).getValue(row, |
| PLong.INSTANCE, ptr); |
| return new MutationState(maxSize, maxSizeBytes, connection) { |
| @Override |
| public long getUpdateCount() { |
| return mutationCount; |
| } |
| }; |
| } finally { |
| iterator.close(); |
| } |
| |
| } |
| |
| @Override |
| public ExplainPlan getExplainPlan() throws SQLException { |
| List<String> queryPlanSteps = aggPlan.getExplainPlan().getPlanSteps(); |
| List<String> planSteps = Lists.newArrayListWithExpectedSize(queryPlanSteps.size()+1); |
| planSteps.add("UPSERT ROWS"); |
| planSteps.addAll(queryPlanSteps); |
| return new ExplainPlan(planSteps); |
| } |
| |
| @Override |
| public Long getEstimatedRowsToScan() throws SQLException { |
| return aggPlan.getEstimatedRowsToScan(); |
| } |
| |
| @Override |
| public Long getEstimatedBytesToScan() throws SQLException { |
| return aggPlan.getEstimatedBytesToScan(); |
| } |
| |
| @Override |
| public Long getEstimateInfoTimestamp() throws SQLException { |
| return aggPlan.getEstimateInfoTimestamp(); |
| } |
| } |
| |
| private class UpsertValuesMutationPlan implements MutationPlan { |
| private final StatementContext context; |
| private final TableRef tableRef; |
| private final int nodeIndexOffset; |
| private final List<Expression> constantExpressions; |
| private final List<PColumn> allColumns; |
| private final int[] columnIndexes; |
| private final Set<PColumn> overlapViewColumns; |
| private final byte[][] values; |
| private final Set<PColumn> addViewColumns; |
| private final PhoenixConnection connection; |
| private final int[] pkSlotIndexes; |
| private final boolean useServerTimestamp; |
| private final byte[] onDupKeyBytes; |
| private final int maxSize; |
| private final int maxSizeBytes; |
| |
| public UpsertValuesMutationPlan(StatementContext context, TableRef tableRef, int nodeIndexOffset, |
| List<Expression> constantExpressions, List<PColumn> allColumns, |
| int[] columnIndexes, Set<PColumn> overlapViewColumns, byte[][] values, |
| Set<PColumn> addViewColumns, PhoenixConnection connection, |
| int[] pkSlotIndexes, boolean useServerTimestamp, byte[] onDupKeyBytes, |
| int maxSize, int maxSizeBytes) { |
| this.context = context; |
| this.tableRef = tableRef; |
| this.nodeIndexOffset = nodeIndexOffset; |
| this.constantExpressions = constantExpressions; |
| this.allColumns = allColumns; |
| this.columnIndexes = columnIndexes; |
| this.overlapViewColumns = overlapViewColumns; |
| this.values = values; |
| this.addViewColumns = addViewColumns; |
| this.connection = connection; |
| this.pkSlotIndexes = pkSlotIndexes; |
| this.useServerTimestamp = useServerTimestamp; |
| this.onDupKeyBytes = onDupKeyBytes; |
| this.maxSize = maxSize; |
| this.maxSizeBytes = maxSizeBytes; |
| } |
| |
| @Override |
| public ParameterMetaData getParameterMetaData() { |
| return context.getBindManager().getParameterMetaData(); |
| } |
| |
| @Override |
| public StatementContext getContext() { |
| return context; |
| } |
| |
| @Override |
| public TableRef getTargetRef() { |
| return tableRef; |
| } |
| |
| @Override |
| public QueryPlan getQueryPlan() { |
| return null; |
| } |
| |
| @Override |
| public Set<TableRef> getSourceRefs() { |
| return Collections.emptySet(); |
| } |
| |
| @Override |
| public Operation getOperation() { |
| return operation; |
| } |
| |
| @Override |
| public MutationState execute() throws SQLException { |
| ImmutableBytesWritable ptr = context.getTempPtr(); |
| final SequenceManager sequenceManager = context.getSequenceManager(); |
| // Next evaluate all the expressions |
| int nodeIndex = nodeIndexOffset; |
| PTable table = tableRef.getTable(); |
| Tuple tuple = sequenceManager.getSequenceCount() == 0 ? null : |
| sequenceManager.newSequenceTuple(null); |
| for (Expression constantExpression : constantExpressions) { |
| PColumn column = allColumns.get(columnIndexes[nodeIndex]); |
| constantExpression.evaluate(tuple, ptr); |
| Object value = null; |
| if (constantExpression.getDataType() != null) { |
| value = constantExpression.getDataType().toObject(ptr, constantExpression.getSortOrder(), |
| constantExpression.getMaxLength(), constantExpression.getScale()); |
| if (!constantExpression.getDataType().isCoercibleTo(column.getDataType(), value)) { |
| throw TypeMismatchException.newException( |
| constantExpression.getDataType(), column.getDataType(), "expression: " |
| + constantExpression.toString() + " in column " + column); |
| } |
| if (!column.getDataType().isSizeCompatible(ptr, value, constantExpression.getDataType(), |
| constantExpression.getSortOrder(), constantExpression.getMaxLength(), |
| constantExpression.getScale(), column.getMaxLength(), column.getScale())) { |
| throw new SQLExceptionInfo.Builder( |
| SQLExceptionCode.DATA_EXCEEDS_MAX_CAPACITY).setColumnName(column.getName().getString()) |
| .setMessage("value=" + constantExpression.toString()).build().buildException(); |
| } |
| } |
| column.getDataType().coerceBytes(ptr, value, constantExpression.getDataType(), |
| constantExpression.getMaxLength(), constantExpression.getScale(), constantExpression.getSortOrder(), |
| column.getMaxLength(), column.getScale(),column.getSortOrder(), |
| table.rowKeyOrderOptimizable()); |
| if (overlapViewColumns.contains(column) && Bytes.compareTo(ptr.get(), ptr.getOffset(), ptr.getLength(), column.getViewConstant(), 0, column.getViewConstant().length-1) != 0) { |
| throw new SQLExceptionInfo.Builder( |
| SQLExceptionCode.CANNOT_UPDATE_VIEW_COLUMN) |
| .setColumnName(column.getName().getString()) |
| .setMessage("value=" + constantExpression.toString()).build().buildException(); |
| } |
| values[nodeIndex] = ByteUtil.copyKeyBytesIfNecessary(ptr); |
| nodeIndex++; |
| } |
| // Add columns based on view |
| for (PColumn column : addViewColumns) { |
| if (IndexUtil.getViewConstantValue(column, ptr)) { |
| values[nodeIndex++] = ByteUtil.copyKeyBytesIfNecessary(ptr); |
| } else { |
| throw new IllegalStateException(); |
| } |
| } |
| MultiRowMutationState mutation = new MultiRowMutationState(1); |
| IndexMaintainer indexMaintainer = null; |
| byte[][] viewConstants = null; |
| if (table.getIndexType() == IndexType.LOCAL) { |
| PTable parentTable = |
| statement |
| .getConnection() |
| .getMetaDataCache() |
| .getTableRef( |
| new PTableKey(statement.getConnection().getTenantId(), |
| table.getParentName().getString())).getTable(); |
| indexMaintainer = table.getIndexMaintainer(parentTable, connection); |
| viewConstants = IndexUtil.getViewConstants(parentTable); |
| } |
| setValues(values, pkSlotIndexes, columnIndexes, table, mutation, statement, useServerTimestamp, indexMaintainer, viewConstants, onDupKeyBytes, 0); |
| return new MutationState(tableRef, mutation, 0, maxSize, maxSizeBytes, connection); |
| } |
| |
| @Override |
| public ExplainPlan getExplainPlan() throws SQLException { |
| List<String> planSteps = Lists.newArrayListWithExpectedSize(2); |
| if (context.getSequenceManager().getSequenceCount() > 0) { |
| planSteps.add("CLIENT RESERVE " + context.getSequenceManager().getSequenceCount() + " SEQUENCES"); |
| } |
| planSteps.add("PUT SINGLE ROW"); |
| return new ExplainPlan(planSteps); |
| } |
| |
| @Override |
| public Long getEstimatedRowsToScan() throws SQLException { |
| return 0l; |
| } |
| |
| @Override |
| public Long getEstimatedBytesToScan() throws SQLException { |
| return 0l; |
| } |
| |
| @Override |
| public Long getEstimateInfoTimestamp() throws SQLException { |
| return 0l; |
| } |
| } |
| |
| private class ClientUpsertSelectMutationPlan implements MutationPlan { |
| private final QueryPlan queryPlan; |
| private final TableRef tableRef; |
| private final QueryPlan originalQueryPlan; |
| private final UpsertingParallelIteratorFactory parallelIteratorFactory; |
| private final RowProjector projector; |
| private final int[] columnIndexes; |
| private final int[] pkSlotIndexes; |
| private final boolean useServerTimestamp; |
| private final int maxSize; |
| private final int maxSizeBytes; |
| |
| public ClientUpsertSelectMutationPlan(QueryPlan queryPlan, TableRef tableRef, QueryPlan originalQueryPlan, UpsertingParallelIteratorFactory parallelIteratorFactory, RowProjector projector, int[] columnIndexes, int[] pkSlotIndexes, boolean useServerTimestamp, int maxSize, int maxSizeBytes) { |
| this.queryPlan = queryPlan; |
| this.tableRef = tableRef; |
| this.originalQueryPlan = originalQueryPlan; |
| this.parallelIteratorFactory = parallelIteratorFactory; |
| this.projector = projector; |
| this.columnIndexes = columnIndexes; |
| this.pkSlotIndexes = pkSlotIndexes; |
| this.useServerTimestamp = useServerTimestamp; |
| this.maxSize = maxSize; |
| this.maxSizeBytes = maxSizeBytes; |
| queryPlan.getContext().setClientSideUpsertSelect(true); |
| } |
| |
| @Override |
| public ParameterMetaData getParameterMetaData() { |
| return queryPlan.getContext().getBindManager().getParameterMetaData(); |
| } |
| |
| @Override |
| public StatementContext getContext() { |
| return queryPlan.getContext(); |
| } |
| |
| @Override |
| public TableRef getTargetRef() { |
| return tableRef; |
| } |
| |
| @Override |
| public QueryPlan getQueryPlan() { |
| return queryPlan; |
| } |
| |
| @Override |
| public Set<TableRef> getSourceRefs() { |
| return originalQueryPlan.getSourceRefs(); |
| } |
| |
| @Override |
| public Operation getOperation() { |
| return operation; |
| } |
| |
| @Override |
| public MutationState execute() throws SQLException { |
| ResultIterator iterator = queryPlan.iterator(); |
| if (parallelIteratorFactory == null) { |
| return upsertSelect(new StatementContext(statement, queryPlan.getContext().getScan()), tableRef, projector, iterator, columnIndexes, pkSlotIndexes, useServerTimestamp, false); |
| } |
| try { |
| parallelIteratorFactory.setRowProjector(projector); |
| parallelIteratorFactory.setColumnIndexes(columnIndexes); |
| parallelIteratorFactory.setPkSlotIndexes(pkSlotIndexes); |
| Tuple tuple; |
| long totalRowCount = 0; |
| StatementContext context = queryPlan.getContext(); |
| while ((tuple=iterator.next()) != null) {// Runs query |
| Cell kv = tuple.getValue(0); |
| totalRowCount += PLong.INSTANCE.getCodec().decodeLong(kv.getValueArray(), kv.getValueOffset(), SortOrder.getDefault()); |
| } |
| // Return total number of rows that have been updated. In the case of auto commit being off |
| // the mutations will all be in the mutation state of the current connection. |
| MutationState mutationState = new MutationState(maxSize, maxSizeBytes, statement.getConnection(), totalRowCount); |
| /* |
| * All the metrics collected for measuring the reads done by the parallel mutating iterators |
| * is included in the ReadMetricHolder of the statement context. Include these metrics in the |
| * returned mutation state so they can be published on commit. |
| */ |
| mutationState.setReadMetricQueue(context.getReadMetricsQueue()); |
| return mutationState; |
| } finally { |
| iterator.close(); |
| } |
| } |
| |
| @Override |
| public ExplainPlan getExplainPlan() throws SQLException { |
| List<String> queryPlanSteps = queryPlan.getExplainPlan().getPlanSteps(); |
| List<String> planSteps = Lists.newArrayListWithExpectedSize(queryPlanSteps.size()+1); |
| planSteps.add("UPSERT SELECT"); |
| planSteps.addAll(queryPlanSteps); |
| return new ExplainPlan(planSteps); |
| } |
| |
| @Override |
| public Long getEstimatedRowsToScan() throws SQLException { |
| return queryPlan.getEstimatedRowsToScan(); |
| } |
| |
| @Override |
| public Long getEstimatedBytesToScan() throws SQLException { |
| return queryPlan.getEstimatedBytesToScan(); |
| } |
| |
| @Override |
| public Long getEstimateInfoTimestamp() throws SQLException { |
| return queryPlan.getEstimateInfoTimestamp(); |
| } |
| } |
| } |
