Set the CQL prepared statement for the output of this job.

@param conf Job configuration you are about to run
@param cql
483351
public static void setOutputCql(Configuration conf, String cql)
    {
        if (cql == null || cql.isEmpty())
            return;
        
        conf.set(OUTPUT_CQL, cql);
    }
1180873
public static void setConf(Object theObject, Configuration conf,
      boolean supportJobConf) {
    if (conf != null) {
      if (theObject instanceof Configurable) {
        ((Configurable) theObject).setConf(conf);
      }
      if (supportJobConf) {
        setJobConf(theObject, conf);
      }
    }
  }
1343325
@Override
    public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException
    {
        checkClosed();
        throw SQLError.noSupport();
    }
418250
protected <T> CompletableFuture<ResultSet> executeQuery(final EntityInfo<T> info, final AsyncConnection conn, final String sql) {
        final byte[] bytes = conn.getAttribute(CONN_ATTR_BYTESBAME);
        final ByteBufferWriter writer = ByteBufferWriter.create(bufferPool);
        {
            writer.put((byte) 'Q');
            int start = writer.position();
            writer.putInt(0);
            writeUTF8String(writer, sql);
            writer.putInt(start, writer.position() - start);
        }
        final ByteBuffer[] buffers = writer.toBuffers();
        final CompletableFuture<ResultSet> future = new CompletableFuture();
        conn.write(buffers, buffers, new CompletionHandler<Integer, ByteBuffer[]>() {
            @Override
            public void completed(Integer result, ByteBuffer[] attachment1) {
                if (result < 0) {
                    failed(new SQLException("Write Buffer Error"), attachment1);
                    return;
                }
                int index = -1;
                for (int i = 0; i < attachment1.length; i++) {
                    if (attachment1[i].hasRemaining()) {
                        index = i;
                        break;
                    }
                    bufferPool.accept(attachment1[i]);
                }
                if (index == 0) {
                    conn.write(attachment1, attachment1, this);
                    return;
                } else if (index > 0) {
                    ByteBuffer[] newattachs = new ByteBuffer[attachment1.length - index];
                    System.arraycopy(attachment1, index, newattachs, 0, newattachs.length);
                    conn.write(newattachs, newattachs, this);
                    return;
                }
                final PgResultSet resultSet = new PgResultSet();
                final List<ByteBuffer> readBuffs = new ArrayList<>();
                conn.read(new CompletionHandler<Integer, ByteBuffer>() {
                    @Override
                    public void completed(Integer result, ByteBuffer attachment2) {
                        if (result < 0) {
                            failed(new SQLException("Read Buffer Error"), attachment2);
                            return;
                        }
                        if (result == 8192 || !attachment2.hasRemaining()) { //postgresql数据包上限为8192 还有数据
                            attachment2.flip();
                            readBuffs.add(attachment2);
                            conn.read(this);
                            return;
                        }
                        attachment2.flip();
                        readBuffs.add(attachment2);
                        final ByteBufferReader buffer = ByteBufferReader.create(readBuffs);
                        boolean endok = false;
                        boolean futureover = false;
                        while (buffer.hasRemaining()) {
                            final char cmd = (char) buffer.get();
                            int length = buffer.getInt();
                            switch (cmd) {
                                case 'E':
                                    byte[] field = new byte[255];
                                    String level = null,
                                     code = null,
                                     message = null;
                                    for (byte type = buffer.get(); type != 0; type = buffer.get()) {
                                        String value = readUTF8String(buffer, field);
                                        if (type == (byte) 'S') {
                                            level = value;
                                        } else if (type == 'C') {
                                            code = value;
                                        } else if (type == 'M') {
                                            message = value;
                                        }
                                    }
                                    future.completeExceptionally(new SQLException(message, code, 0));
                                    futureover = true;
                                    break;
                                case 'T':
                                    RowDesc rowDesc = new RespRowDescDecoder().read(buffer, length, bytes);
                                    resultSet.setRowDesc(rowDesc);
                                    break;
                                case 'D':
                                    RowData rowData = new RespRowDataDecoder().read(buffer, length, bytes);
                                    resultSet.addRowData(rowData);
                                    futureover = true;
                                    break;
                                case 'Z':
                                    //buffer.position(buffer.position() + length - 4);
                                    buffer.skip(length - 4);
                                    endok = true;
                                    break;
                                default:
                                    //buffer.position(buffer.position() + length - 4);
                                    buffer.skip(length - 4);
                                    break;
                            }
                        }
                        for (ByteBuffer buf : readBuffs) {
                            bufferPool.accept(buf);
                        }
                        if (!futureover) future.completeExceptionally(new SQLException("SQL(" + sql + ") executeQuery error"));
                        if (endok) {
                            readPool.offerConnection(conn);
                            future.complete(resultSet);
                        } else {
                            conn.dispose();
                        }
                    }

                    @Override
                    public void failed(Throwable exc, ByteBuffer attachment2) {
                        //不用bufferPool.accept
                        future.completeExceptionally(exc);
                        conn.dispose();
                    }
                });
            }

            @Override
            public void failed(Throwable exc, ByteBuffer[] attachment1) {
                for (ByteBuffer attach : attachment1) {
                    bufferPool.accept(attach);
                }
                future.completeExceptionally(exc);
            }
        });
        return future;
    }
495575
private static void setIfUnset(JobConf conf, String key, String value) {
    if (conf.get(key) == null) {
      conf.set(key, value);
    }
  }
1194707
public PreparedStatement prepareStatement(final String sql) throws SQLException {
    return internalPrepareStatement(sql,
        ResultSet.TYPE_FORWARD_ONLY,
        ResultSet.CONCUR_READ_ONLY,
        Statement.NO_GENERATED_KEYS);
  }
23565
@NonNull
  public static CqlIdentifier fromCql(@NonNull String cql) {
    Preconditions.checkNotNull(cql, "cql must not be null");
    final String internal;
    if (Strings.isDoubleQuoted(cql)) {
      internal = Strings.unDoubleQuote(cql);
    } else {
      internal = cql.toLowerCase();
      Preconditions.checkArgument(
          !Strings.needsDoubleQuotes(internal), "Invalid CQL form [%s]: needs double quotes", cql);
    }
    return fromInternal(internal);
  }
911355
public CqlQuery<K, N, V> setQuery(String query) {
    this.query = StringSerializer.get().toByteBuffer(query);
    return this;
  }
493638
public static void configureBigQueryInput(
      Configuration config, String fullyQualifiedInputTableId)
      throws IOException {
    TableReference parsedTable = BigQueryStrings.parseTableReference(fullyQualifiedInputTableId);
    configureBigQueryInput(
        config, parsedTable.getProjectId(), parsedTable.getDatasetId(), parsedTable.getTableId());
  }
172922
protected <T> void prepareCriteria(Query<T> jpaQuery) {
        if (cacheQueries) {
            jpaQuery.setCacheable(true);
        }
        if (shouldPassReadOnlyToHibernate()) {
            jpaQuery.setReadOnly(true);
        }
        SessionHolder sessionHolder = (SessionHolder) TransactionSynchronizationManager.getResource(sessionFactory);
        if (sessionHolder != null && sessionHolder.hasTimeout()) {
            jpaQuery.setTimeout(sessionHolder.getTimeToLiveInSeconds());
        }
    }
1194111
private void sendCqlQuery(Map<String, ByteBuffer> key, List<ByteBuffer> bindedVariables) throws IOException
    {
        try
        {
            writer.write(key, bindedVariables);
        }
        catch (InterruptedException e)
        {
            throw new IOException(e);
        }
    }
73441
public <T> T execute(final String query, Object connection) {
        try {
            org.apache.cassandra.thrift.Cassandra.Client conn =
                (org.apache.cassandra.thrift.Cassandra.Client) connection;
            conn.set_cql_version(CassandraConstants.CQL_VERSION_3_0);
            KunderaCoreUtils.printQuery(query, showQuery);
            return (T) conn.execute_cql3_query(ByteBuffer.wrap(query.getBytes(Constants.CHARSET_UTF8)),
                Compression.NONE, getConsistencyLevel());
        } catch (Exception e) {
            log.error("Error while executing query {}", query);
            throw new KunderaException(e);
        }
    }
671629
private PreparedStatement prepareUpdate(String tableName, Update update) {
        StringBuilder cql = new StringBuilder();
        switch (update) {
        case INSERT_ROW:
            // INSERT INTO <keyspace>.<table> (key,column1,value) VALUES (<key>, <colname>, <colvalue>);
            cql.append("INSERT INTO ");
            cql.append(m_keyspace);
            cql.append(".");
            cql.append(tableName);
            cql.append(" (key,column1,value) VALUES (?, ?, ?);");
            break;
        case INSERT_ROW_TS:
            // INSERT INTO <keyspace>.<table> (key,column1,value) VALUES (<key>, <colname>, <colvalue>) USING TIMESTAMP <timestamp>;
            // Note timestamp is the last parameter
            cql.append("INSERT INTO ");
            cql.append(m_keyspace);
            cql.append(".");
            cql.append(tableName);
            cql.append(" (key,column1,value) VALUES (?, ?, ?) USING TIMESTAMP ?;");
            break;
        case DELETE_COLUMN:
            // DELETE FROM <keyspace>.<table> WHERE key=<key> AND column1=<column name>;
            cql.append("DELETE FROM ");
            cql.append(m_keyspace);
            cql.append(".");
            cql.append(tableName);
            cql.append(" WHERE key=? AND column1=?;");
            break;
        case DELETE_COLUMN_TS:
            // DELETE FROM <keyspace>.<table> USING TIMESTAMP <timestamp> WHERE key=<key> AND column1=<column name>;
            // Note timestamp is the first parameter
            cql.append("DELETE FROM ");
            cql.append(m_keyspace);
            cql.append(".");
            cql.append(tableName);
            cql.append(" USING TIMESTAMP ? WHERE key=? AND column1=?;");
            break;
        case DELETE_ROW:
            // DELETE FROM <keyspace>.<table> WHERE key='key';
            cql.append("DELETE FROM ");
            cql.append(m_keyspace);
            cql.append(".");
            cql.append(tableName);
            cql.append(" WHERE key=?;");
            break;
        case DELETE_ROW_TS:
            // DELETE FROM <keyspace>.<table> USING TIMESTAMP <timestamp> WHERE key='key';
            // Note timestamp is the first parameter
            cql.append("DELETE FROM ");
            cql.append(m_keyspace);
            cql.append(".");
            cql.append(tableName);
            cql.append(" USING TIMESTAMP ? WHERE key=?;");
            break;
        }
        m_logger.debug("Preparing update statement: {}", cql);
        return ((CQLService)DBService.instance(m_tenant)).getSession().prepare(cql.toString());
    }
57476
public void prepareForExecution(JobContextImpl jobContext, String restartOn) {
		this.jobContext = jobContext;
		this.jobNavigator = jobContext.getNavigator();
		jobContext.setExecutionId(executionId);
		jobContext.setInstanceId(jobInstance.getInstanceId());
		this.restartOn = restartOn;
		operatorJobExecution.setJobContext(jobContext);
	}
580289
public static CFMetaData parseCreateStatement(String cql) throws RequestValidationException {
        final CreateColumnFamilyStatement statement =
                (CreateColumnFamilyStatement) QueryProcessor.parseStatement(cql).prepare().statement;

        final CFMetaData cfm =
                new CFMetaData("assess", "kvs_strict", ColumnFamilyType.Standard, statement.comparator, null);

        statement.applyPropertiesTo(cfm);
        return cfm;
    }
857478
@Override
    public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException
    {
        checkClosed();
        throw SQLError.noSupport();
    }
63035
public static void setOutput(JobConf job, String tableName, String... fieldNames) {
    job.setOutputFormat(DBOutputFormat.class);
    job.setReduceSpeculativeExecution(false);

    DBConfiguration dbConf = new DBConfiguration(job);
    
    dbConf.setOutputTableName(tableName);
    dbConf.setOutputFieldNames(fieldNames);
  }