http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/it/java/org/apache/phoenix/tx/TransactionIT.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/it/java/org/apache/phoenix/tx/TransactionIT.java b/phoenix-core/src/it/java/org/apache/phoenix/tx/TransactionIT.java index 1399f6c..f37d09b 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/tx/TransactionIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/tx/TransactionIT.java @@ -26,337 +26,110 @@ import static org.junit.Assert.fail; import java.sql.Connection; import java.sql.DatabaseMetaData; -import java.sql.Date; import java.sql.DriverManager; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; -import java.util.List; import java.util.Properties; import org.apache.hadoop.hbase.HColumnDescriptor; import org.apache.hadoop.hbase.HTableDescriptor; -import org.apache.hadoop.hbase.TableName; -import org.apache.hadoop.hbase.client.HBaseAdmin; -import org.apache.hadoop.hbase.client.HTableInterface; -import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.util.Bytes; -import org.apache.phoenix.coprocessor.PhoenixTransactionalProcessor; import org.apache.phoenix.end2end.ParallelStatsDisabledIT; import org.apache.phoenix.exception.SQLExceptionCode; import org.apache.phoenix.jdbc.PhoenixConnection; import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData; import org.apache.phoenix.query.QueryConstants; -import org.apache.phoenix.query.QueryServices; import org.apache.phoenix.query.QueryServicesOptions; -import org.apache.phoenix.schema.PTable; import org.apache.phoenix.schema.PTableKey; -import org.apache.phoenix.schema.types.PInteger; -import org.apache.phoenix.util.ByteUtil; import org.apache.phoenix.util.PropertiesUtil; import org.apache.phoenix.util.StringUtil; import org.apache.phoenix.util.TestUtil; import org.apache.tephra.TxConstants; -import org.junit.Ignore; import org.junit.Test; -import com.google.common.collect.Lists; +public class TransactionIT extends ParallelStatsDisabledIT { -public class TransactionIT extends ParallelStatsDisabledIT { - @Test - public void testReadOwnWrites() throws Exception { - String transTableName = generateUniqueName(); - String fullTableName = INDEX_DATA_SCHEMA + QueryConstants.NAME_SEPARATOR + transTableName; - String selectSql = "SELECT * FROM "+ fullTableName; - try (Connection conn = DriverManager.getConnection(getUrl())) { - TestUtil.createTransactionalTable(conn, fullTableName); - conn.setAutoCommit(false); - ResultSet rs = conn.createStatement().executeQuery(selectSql); - assertFalse(rs.next()); - - String upsert = "UPSERT INTO " + fullTableName + "(varchar_pk, char_pk, int_pk, long_pk, decimal_pk, date_pk) VALUES(?, ?, ?, ?, ?, ?)"; - PreparedStatement stmt = conn.prepareStatement(upsert); - // upsert two rows - TestUtil.setRowKeyColumns(stmt, 1); - stmt.execute(); - TestUtil.setRowKeyColumns(stmt, 2); - stmt.execute(); - - // verify rows can be read even though commit has not been called - rs = conn.createStatement().executeQuery(selectSql); - TestUtil.validateRowKeyColumns(rs, 1); - TestUtil.validateRowKeyColumns(rs, 2); - assertFalse(rs.next()); - - conn.commit(); - - // verify rows can be read after commit - rs = conn.createStatement().executeQuery(selectSql); - TestUtil.validateRowKeyColumns(rs, 1); - TestUtil.validateRowKeyColumns(rs, 2); - assertFalse(rs.next()); - } + public void testReCreateTxnTableAfterDroppingExistingNonTxnTable() throws SQLException { + String tableName = generateUniqueName(); + Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES); + Connection conn = DriverManager.getConnection(getUrl(), props); + conn.setAutoCommit(false); + Statement stmt = conn.createStatement(); + stmt.execute("CREATE TABLE " + tableName + "(k VARCHAR PRIMARY KEY, v1 VARCHAR, v2 VARCHAR)"); + stmt.execute("DROP TABLE " + tableName); + stmt.execute("CREATE TABLE " + tableName + "(k VARCHAR PRIMARY KEY, v1 VARCHAR, v2 VARCHAR) TRANSACTIONAL=true"); + stmt.execute("CREATE INDEX " + tableName + "_IDX ON " + tableName + " (v1) INCLUDE(v2)"); + assertTrue(conn.unwrap(PhoenixConnection.class).getTable(new PTableKey(null, tableName)).isTransactional()); + assertTrue(conn.unwrap(PhoenixConnection.class).getTable(new PTableKey(null, tableName + "_IDX")).isTransactional()); } @Test - public void testTxnClosedCorrecty() throws Exception { - String transTableName = generateUniqueName(); - String fullTableName = INDEX_DATA_SCHEMA + QueryConstants.NAME_SEPARATOR + transTableName; - String selectSql = "SELECT * FROM "+fullTableName; - try (Connection conn = DriverManager.getConnection(getUrl())) { - TestUtil.createTransactionalTable(conn, fullTableName); + public void testRowTimestampDisabled() throws SQLException { + String tableName = generateUniqueName(); + Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES); + try (Connection conn = DriverManager.getConnection(getUrl(), props)) { conn.setAutoCommit(false); - ResultSet rs = conn.createStatement().executeQuery(selectSql); - assertFalse(rs.next()); - - String upsert = "UPSERT INTO " + fullTableName + "(varchar_pk, char_pk, int_pk, long_pk, decimal_pk, date_pk) VALUES(?, ?, ?, ?, ?, ?)"; - PreparedStatement stmt = conn.prepareStatement(upsert); - // upsert two rows - TestUtil.setRowKeyColumns(stmt, 1); - stmt.execute(); - TestUtil.setRowKeyColumns(stmt, 2); - stmt.execute(); - - // verify rows can be read even though commit has not been called - rs = conn.createStatement().executeQuery(selectSql); - TestUtil.validateRowKeyColumns(rs, 1); - TestUtil.validateRowKeyColumns(rs, 2); - // Long currentTx = rs.unwrap(PhoenixResultSet.class).getCurrentRow().getValue(0).getTimestamp(); - assertFalse(rs.next()); - - conn.close(); - // start new connection - // conn.createStatement().executeQuery(selectSql); - // assertFalse("This transaction should not be on the invalid transactions", - // txManager.getCurrentState().getInvalid().contains(currentTx)); - } - } - - @Test - public void testAutoCommitQuerySingleTable() throws Exception { - String transTableName = generateUniqueName(); - String fullTableName = INDEX_DATA_SCHEMA + QueryConstants.NAME_SEPARATOR + transTableName; - try (Connection conn = DriverManager.getConnection(getUrl())) { - TestUtil.createTransactionalTable(conn, fullTableName); - conn.setAutoCommit(true); - // verify no rows returned - ResultSet rs = conn.createStatement().executeQuery("SELECT * FROM " + fullTableName); - assertFalse(rs.next()); - } - } - - @Test - public void testAutoCommitQueryMultiTables() throws Exception { - String transTableName = generateUniqueName(); - String fullTableName = INDEX_DATA_SCHEMA + QueryConstants.NAME_SEPARATOR + transTableName; - try (Connection conn = DriverManager.getConnection(getUrl())) { - TestUtil.createTransactionalTable(conn, fullTableName); - conn.setAutoCommit(true); - // verify no rows returned - ResultSet rs = conn.createStatement().executeQuery("SELECT * FROM " + fullTableName + " a JOIN " + fullTableName + " b ON (a.long_pk = b.int_pk)"); - assertFalse(rs.next()); - } - } - - @Test - public void testColConflicts() throws Exception { - String transTableName = generateUniqueName(); - String fullTableName = INDEX_DATA_SCHEMA + QueryConstants.NAME_SEPARATOR + transTableName; - try (Connection conn1 = DriverManager.getConnection(getUrl()); - Connection conn2 = DriverManager.getConnection(getUrl())) { - TestUtil.createTransactionalTable(conn1, fullTableName); - conn1.setAutoCommit(false); - conn2.setAutoCommit(false); - String selectSql = "SELECT * FROM "+fullTableName; - conn1.setAutoCommit(false); - ResultSet rs = conn1.createStatement().executeQuery(selectSql); - assertFalse(rs.next()); - // upsert row using conn1 - String upsertSql = "UPSERT INTO " + fullTableName + "(varchar_pk, char_pk, int_pk, long_pk, decimal_pk, date_pk, a.int_col1) VALUES(?, ?, ?, ?, ?, ?, ?)"; - PreparedStatement stmt = conn1.prepareStatement(upsertSql); - TestUtil.setRowKeyColumns(stmt, 1); - stmt.setInt(7, 10); - stmt.execute(); - // upsert row using conn2 - stmt = conn2.prepareStatement(upsertSql); - TestUtil.setRowKeyColumns(stmt, 1); - stmt.setInt(7, 11); - stmt.execute(); - - conn1.commit(); - //second commit should fail + Statement stmt = conn.createStatement(); try { - conn2.commit(); + stmt.execute("CREATE TABLE " + tableName + "(k VARCHAR, v VARCHAR, d DATE NOT NULL, CONSTRAINT PK PRIMARY KEY(k,d ROW_TIMESTAMP)) TRANSACTIONAL=true"); fail(); - } - catch (SQLException e) { - assertEquals(e.getErrorCode(), SQLExceptionCode.TRANSACTION_CONFLICT_EXCEPTION.getErrorCode()); } - } - } - - private void testRowConflicts(String fullTableName) throws Exception { - try (Connection conn1 = DriverManager.getConnection(getUrl()); - Connection conn2 = DriverManager.getConnection(getUrl())) { - conn1.setAutoCommit(false); - conn2.setAutoCommit(false); - String selectSql = "SELECT * FROM "+fullTableName; - conn1.setAutoCommit(false); - ResultSet rs = conn1.createStatement().executeQuery(selectSql); - boolean immutableRows = conn1.unwrap(PhoenixConnection.class).getTable(new PTableKey(null, fullTableName)).isImmutableRows(); - assertFalse(rs.next()); - // upsert row using conn1 - String upsertSql = "UPSERT INTO " + fullTableName + "(varchar_pk, char_pk, int_pk, long_pk, decimal_pk, date_pk, a.int_col1) VALUES(?, ?, ?, ?, ?, ?, ?)"; - PreparedStatement stmt = conn1.prepareStatement(upsertSql); - TestUtil.setRowKeyColumns(stmt, 1); - stmt.setInt(7, 10); - stmt.execute(); - // upsert row using conn2 - upsertSql = "UPSERT INTO " + fullTableName + "(varchar_pk, char_pk, int_pk, long_pk, decimal_pk, date_pk, b.int_col2) VALUES(?, ?, ?, ?, ?, ?, ?)"; - stmt = conn2.prepareStatement(upsertSql); - TestUtil.setRowKeyColumns(stmt, 1); - stmt.setInt(7, 11); - stmt.execute(); - - conn1.commit(); - //second commit should fail + catch(SQLException e) { + assertEquals(SQLExceptionCode.CANNOT_CREATE_TXN_TABLE_WITH_ROW_TIMESTAMP.getErrorCode(), e.getErrorCode()); + } + stmt.execute("CREATE TABLE " + tableName + "(k VARCHAR, v VARCHAR, d DATE NOT NULL, CONSTRAINT PK PRIMARY KEY(k,d ROW_TIMESTAMP))"); try { - conn2.commit(); - if (!immutableRows) fail(); - } - catch (SQLException e) { - if (immutableRows) fail(); - assertEquals(e.getErrorCode(), SQLExceptionCode.TRANSACTION_CONFLICT_EXCEPTION.getErrorCode()); + stmt.execute("ALTER TABLE " + tableName + " SET TRANSACTIONAL=true"); + fail(); + } + catch(SQLException e) { + assertEquals(SQLExceptionCode.CANNOT_ALTER_TO_BE_TXN_WITH_ROW_TIMESTAMP.getErrorCode(), e.getErrorCode()); } } } @Test - public void testRowConflictDetected() throws Exception { - String transTableName = generateUniqueName(); - String fullTableName = INDEX_DATA_SCHEMA + QueryConstants.NAME_SEPARATOR + transTableName; - Connection conn = DriverManager.getConnection(getUrl()); - TestUtil.createTransactionalTable(conn, fullTableName); - testRowConflicts(fullTableName); - } - - @Test - public void testNoConflictDetectionForImmutableRows() throws Exception { - String transTableName = generateUniqueName(); - String fullTableName = INDEX_DATA_SCHEMA + QueryConstants.NAME_SEPARATOR + transTableName; - Connection conn = DriverManager.getConnection(getUrl()); - TestUtil.createTransactionalTable(conn, fullTableName); - conn.createStatement().execute("ALTER TABLE " + fullTableName + " SET IMMUTABLE_ROWS=true"); - testRowConflicts(fullTableName); - } - - @Test - public void testNonTxToTxTable() throws Exception { - String nonTxTableName = generateUniqueName(); - String indexName = generateUniqueName() + "_IDX"; + public void testTransactionalTableMetadata() throws SQLException { - Connection conn = DriverManager.getConnection(getUrl()); - conn.createStatement().execute("CREATE TABLE " + nonTxTableName + "(k INTEGER PRIMARY KEY, v VARCHAR)"); - conn.createStatement().execute("UPSERT INTO " + nonTxTableName + " VALUES (1)"); - conn.createStatement().execute("UPSERT INTO " + nonTxTableName + " VALUES (2, 'a')"); - conn.createStatement().execute("UPSERT INTO " + nonTxTableName + " VALUES (3, 'b')"); - conn.commit(); - - conn.createStatement().execute("CREATE INDEX " + indexName + " ON " + nonTxTableName + "(v)"); - // Reset empty column value to an empty value like it is pre-transactions - HTableInterface htable = conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes( nonTxTableName)); - List<Put>puts = Lists.newArrayList(new Put(PInteger.INSTANCE.toBytes(1)), new Put(PInteger.INSTANCE.toBytes(2)), new Put(PInteger.INSTANCE.toBytes(3))); - for (Put put : puts) { - put.add(QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, ByteUtil.EMPTY_BYTE_ARRAY); - } - htable.put(puts); - - conn.createStatement().execute("ALTER TABLE " + nonTxTableName + " SET TRANSACTIONAL=true"); - - htable = conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes( nonTxTableName)); - assertTrue(htable.getTableDescriptor().getCoprocessors().contains(PhoenixTransactionalProcessor.class.getName())); - htable = conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(indexName)); - assertTrue(htable.getTableDescriptor().getCoprocessors().contains(PhoenixTransactionalProcessor.class.getName())); + try (Connection conn = DriverManager.getConnection(getUrl())) { + String transactTableName = generateUniqueName(); + Statement stmt = conn.createStatement(); + stmt.execute("CREATE TABLE " + transactTableName + " (k VARCHAR PRIMARY KEY, v1 VARCHAR, v2 VARCHAR) " + + "TRANSACTIONAL=true"); + conn.commit(); - conn.createStatement().execute("UPSERT INTO " + nonTxTableName + " VALUES (4, 'c')"); - ResultSet rs = conn.createStatement().executeQuery("SELECT /*+ NO_INDEX */ k FROM " + nonTxTableName + " WHERE v IS NULL"); - assertTrue(conn.unwrap(PhoenixConnection.class).getTable(new PTableKey(null, nonTxTableName)).isTransactional()); - assertTrue(rs.next()); - assertEquals(1,rs.getInt(1)); - assertFalse(rs.next()); - conn.commit(); - - conn.createStatement().execute("UPSERT INTO " + nonTxTableName + " VALUES (5, 'd')"); - rs = conn.createStatement().executeQuery("SELECT k FROM " + nonTxTableName); - assertTrue(conn.unwrap(PhoenixConnection.class).getTable(new PTableKey(null, indexName)).isTransactional()); - assertTrue(rs.next()); - assertEquals(1,rs.getInt(1)); - assertTrue(rs.next()); - assertEquals(2,rs.getInt(1)); - assertTrue(rs.next()); - assertEquals(3,rs.getInt(1)); - assertTrue(rs.next()); - assertEquals(4,rs.getInt(1)); - assertTrue(rs.next()); - assertEquals(5,rs.getInt(1)); - assertFalse(rs.next()); - conn.rollback(); - - rs = conn.createStatement().executeQuery("SELECT k FROM " + nonTxTableName); - assertTrue(rs.next()); - assertEquals(1,rs.getInt(1)); - assertTrue(rs.next()); - assertEquals(2,rs.getInt(1)); - assertTrue(rs.next()); - assertEquals(3,rs.getInt(1)); - assertTrue(rs.next()); - assertEquals(4,rs.getInt(1)); - assertFalse(rs.next()); + DatabaseMetaData dbmd = conn.getMetaData(); + ResultSet rs = dbmd.getTables(null, null, StringUtil.escapeLike(transactTableName), null); + assertTrue(rs.next()); + assertEquals("Transactional table was not marked as transactional in JDBC API.", + "true", rs.getString(PhoenixDatabaseMetaData.TRANSACTIONAL)); + + String nonTransactTableName = generateUniqueName(); + Statement stmt2 = conn.createStatement(); + stmt2.execute("CREATE TABLE " + nonTransactTableName + "(k VARCHAR PRIMARY KEY, v1 VARCHAR, v2 VARCHAR) "); + conn.commit(); + + ResultSet rs2 = dbmd.getTables(null, null, StringUtil.escapeLike(nonTransactTableName), null); + assertTrue(rs2.next()); + assertEquals("Non-transactional table was marked as transactional in JDBC API.", + "false", rs2.getString(PhoenixDatabaseMetaData.TRANSACTIONAL)); + } } - @Ignore @Test - public void testNonTxToTxTableFailure() throws Exception { - String nonTxTableName = generateUniqueName(); - - Connection conn = DriverManager.getConnection(getUrl()); - // Put table in SYSTEM schema to prevent attempts to update the cache after we disable SYSTEM.CATALOG - conn.createStatement().execute("CREATE TABLE SYSTEM." + nonTxTableName + "(k INTEGER PRIMARY KEY, v VARCHAR)"); - conn.createStatement().execute("UPSERT INTO SYSTEM." + nonTxTableName + " VALUES (1)"); - conn.commit(); - // Reset empty column value to an empty value like it is pre-transactions - HTableInterface htable = conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes("SYSTEM." + nonTxTableName)); - Put put = new Put(PInteger.INSTANCE.toBytes(1)); - put.add(QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, ByteUtil.EMPTY_BYTE_ARRAY); - htable.put(put); - - HBaseAdmin admin = conn.unwrap(PhoenixConnection.class).getQueryServices().getAdmin(); - admin.disableTable(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME); - try { - // This will succeed initially in updating the HBase metadata, but then will fail when - // the SYSTEM.CATALOG table is attempted to be updated, exercising the code to restore - // the coprocessors back to the non transactional ones. - conn.createStatement().execute("ALTER TABLE SYSTEM." + nonTxTableName + " SET TRANSACTIONAL=true"); + public void testOnDupKeyForTransactionalTable() throws Exception { + // TODO: we should support having a transactional table defined for a connectionless connection + try (Connection conn = DriverManager.getConnection(getUrl())) { + String transactTableName = generateUniqueName(); + conn.createStatement().execute("CREATE TABLE " + transactTableName + " (k integer not null primary key, v bigint) TRANSACTIONAL=true"); + conn.createStatement().execute("UPSERT INTO " + transactTableName + " VALUES(0,0) ON DUPLICATE KEY UPDATE v = v + 1"); fail(); } catch (SQLException e) { - assertTrue(e.getMessage().contains(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME + " is disabled")); - } finally { - admin.enableTable(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME); - admin.close(); + assertEquals(SQLExceptionCode.CANNOT_USE_ON_DUP_KEY_FOR_TRANSACTIONAL.getErrorCode(), e.getErrorCode()); } - - ResultSet rs = conn.createStatement().executeQuery("SELECT k FROM SYSTEM." + nonTxTableName + " WHERE v IS NULL"); - assertTrue(rs.next()); - assertEquals(1,rs.getInt(1)); - assertFalse(rs.next()); - - htable = conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes("SYSTEM." + nonTxTableName)); - assertFalse(htable.getTableDescriptor().getCoprocessors().contains(PhoenixTransactionalProcessor.class.getName())); - assertEquals(1,conn.unwrap(PhoenixConnection.class).getQueryServices(). - getTableDescriptor(Bytes.toBytes("SYSTEM." + nonTxTableName)). - getFamily(QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES).getMaxVersions()); } @Test @@ -439,120 +212,38 @@ public class TransactionIT extends ParallelStatsDisabledIT { } @Test - public void testCreateTableToBeTransactional() throws Exception { - Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES); - Connection conn = DriverManager.getConnection(getUrl(), props); - String t1 = generateUniqueName(); - String t2 = generateUniqueName(); - String ddl = "CREATE TABLE " + t1 + " (k varchar primary key) transactional=true"; - conn.createStatement().execute(ddl); - PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); - PTable table = pconn.getTable(new PTableKey(null, t1)); - HTableInterface htable = pconn.getQueryServices().getTable(Bytes.toBytes(t1)); - assertTrue(table.isTransactional()); - assertTrue(htable.getTableDescriptor().getCoprocessors().contains(PhoenixTransactionalProcessor.class.getName())); - - try { - ddl = "ALTER TABLE " + t1 + " SET transactional=false"; - conn.createStatement().execute(ddl); - fail(); - } catch (SQLException e) { - assertEquals(SQLExceptionCode.TX_MAY_NOT_SWITCH_TO_NON_TX.getErrorCode(), e.getErrorCode()); - } - - HBaseAdmin admin = pconn.getQueryServices().getAdmin(); - HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(t2)); - desc.addFamily(new HColumnDescriptor(QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES)); - admin.createTable(desc); - ddl = "CREATE TABLE " + t2 + " (k varchar primary key) transactional=true"; - conn.createStatement().execute(ddl); - assertEquals(Boolean.TRUE.toString(), admin.getTableDescriptor(TableName.valueOf(t2)).getValue(TxConstants.READ_NON_TX_DATA)); - - // Should be ok, as HBase metadata should match existing metadata. - ddl = "CREATE TABLE IF NOT EXISTS " + t1 + " (k varchar primary key)"; - try { - conn.createStatement().execute(ddl); - fail(); - } catch (SQLException e) { - assertEquals(SQLExceptionCode.TX_MAY_NOT_SWITCH_TO_NON_TX.getErrorCode(), e.getErrorCode()); - } - ddl += " transactional=true"; - conn.createStatement().execute(ddl); - table = pconn.getTable(new PTableKey(null, t1)); - htable = pconn.getQueryServices().getTable(Bytes.toBytes(t1)); - assertTrue(table.isTransactional()); - assertTrue(htable.getTableDescriptor().getCoprocessors().contains(PhoenixTransactionalProcessor.class.getName())); - } - - @Test - public void testCurrentDate() throws Exception { + public void testColConflicts() throws Exception { String transTableName = generateUniqueName(); String fullTableName = INDEX_DATA_SCHEMA + QueryConstants.NAME_SEPARATOR + transTableName; - String selectSql = "SELECT current_date() FROM "+fullTableName; - try (Connection conn = DriverManager.getConnection(getUrl())) { - TestUtil.createTransactionalTable(conn, fullTableName); - conn.setAutoCommit(false); - ResultSet rs = conn.createStatement().executeQuery(selectSql); + try (Connection conn1 = DriverManager.getConnection(getUrl()); + Connection conn2 = DriverManager.getConnection(getUrl())) { + TestUtil.createTransactionalTable(conn1, fullTableName); + conn1.setAutoCommit(false); + conn2.setAutoCommit(false); + String selectSql = "SELECT * FROM "+fullTableName; + conn1.setAutoCommit(false); + ResultSet rs = conn1.createStatement().executeQuery(selectSql); assertFalse(rs.next()); - - String upsert = "UPSERT INTO " + fullTableName + "(varchar_pk, char_pk, int_pk, long_pk, decimal_pk, date_pk) VALUES(?, ?, ?, ?, ?, ?)"; - PreparedStatement stmt = conn.prepareStatement(upsert); - // upsert two rows + // upsert row using conn1 + String upsertSql = "UPSERT INTO " + fullTableName + "(varchar_pk, char_pk, int_pk, long_pk, decimal_pk, date_pk, a.int_col1) VALUES(?, ?, ?, ?, ?, ?, ?)"; + PreparedStatement stmt = conn1.prepareStatement(upsertSql); TestUtil.setRowKeyColumns(stmt, 1); + stmt.setInt(7, 10); + stmt.execute(); + // upsert row using conn2 + stmt = conn2.prepareStatement(upsertSql); + TestUtil.setRowKeyColumns(stmt, 1); + stmt.setInt(7, 11); stmt.execute(); - conn.commit(); - - rs = conn.createStatement().executeQuery(selectSql); - assertTrue(rs.next()); - Date date1 = rs.getDate(1); - assertFalse(rs.next()); - - Thread.sleep(1000); - rs = conn.createStatement().executeQuery(selectSql); - assertTrue(rs.next()); - Date date2 = rs.getDate(1); - assertFalse(rs.next()); - assertTrue("current_date() should change while executing multiple statements", date2.getTime() > date1.getTime()); - } - } - - @Test - public void testReCreateTxnTableAfterDroppingExistingNonTxnTable() throws SQLException { - String tableName = generateUniqueName(); - Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES); - Connection conn = DriverManager.getConnection(getUrl(), props); - conn.setAutoCommit(false); - Statement stmt = conn.createStatement(); - stmt.execute("CREATE TABLE " + tableName + "(k VARCHAR PRIMARY KEY, v1 VARCHAR, v2 VARCHAR)"); - stmt.execute("DROP TABLE " + tableName); - stmt.execute("CREATE TABLE " + tableName + "(k VARCHAR PRIMARY KEY, v1 VARCHAR, v2 VARCHAR) TRANSACTIONAL=true"); - stmt.execute("CREATE INDEX " + tableName + "_IDX ON " + tableName + " (v1) INCLUDE(v2)"); - assertTrue(conn.unwrap(PhoenixConnection.class).getTable(new PTableKey(null, tableName)).isTransactional()); - assertTrue(conn.unwrap(PhoenixConnection.class).getTable(new PTableKey(null, tableName + "_IDX")).isTransactional()); - } - - @Test - public void testRowTimestampDisabled() throws SQLException { - String tableName = generateUniqueName(); - Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES); - try (Connection conn = DriverManager.getConnection(getUrl(), props)) { - conn.setAutoCommit(false); - Statement stmt = conn.createStatement(); - try { - stmt.execute("CREATE TABLE " + tableName + "(k VARCHAR, v VARCHAR, d DATE NOT NULL, CONSTRAINT PK PRIMARY KEY(k,d ROW_TIMESTAMP)) TRANSACTIONAL=true"); - fail(); - } - catch(SQLException e) { - assertEquals(SQLExceptionCode.CANNOT_CREATE_TXN_TABLE_WITH_ROW_TIMESTAMP.getErrorCode(), e.getErrorCode()); - } - stmt.execute("CREATE TABLE " + tableName + "(k VARCHAR, v VARCHAR, d DATE NOT NULL, CONSTRAINT PK PRIMARY KEY(k,d ROW_TIMESTAMP))"); + conn1.commit(); + //second commit should fail try { - stmt.execute("ALTER TABLE " + tableName + " SET TRANSACTIONAL=true"); + conn2.commit(); fail(); - } - catch(SQLException e) { - assertEquals(SQLExceptionCode.CANNOT_ALTER_TO_BE_TXN_WITH_ROW_TIMESTAMP.getErrorCode(), e.getErrorCode()); + } + catch (SQLException e) { + assertEquals(e.getErrorCode(), SQLExceptionCode.TRANSACTION_CONFLICT_EXCEPTION.getErrorCode()); } } } @@ -600,118 +291,4 @@ public class TransactionIT extends ParallelStatsDisabledIT { conn.close(); } } - - @Test - public void testParallelUpsertSelect() throws Exception { - Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES); - props.setProperty(QueryServices.MUTATE_BATCH_SIZE_BYTES_ATTRIB, Integer.toString(512)); - props.setProperty(QueryServices.SCAN_CACHE_SIZE_ATTRIB, Integer.toString(3)); - props.setProperty(QueryServices.SCAN_RESULT_CHUNK_SIZE, Integer.toString(3)); - Connection conn = DriverManager.getConnection(getUrl(), props); - conn.setAutoCommit(false); - String fullTableName1 = generateUniqueName(); - String fullTableName2 = generateUniqueName(); - String sequenceName = "S_" + generateUniqueName(); - conn.createStatement().execute("CREATE SEQUENCE " + sequenceName); - conn.createStatement().execute("CREATE TABLE " + fullTableName1 + " (pk INTEGER PRIMARY KEY, val INTEGER) SALT_BUCKETS=4,TRANSACTIONAL=true"); - conn.createStatement().execute("CREATE TABLE " + fullTableName2 + " (pk INTEGER PRIMARY KEY, val INTEGER) TRANSACTIONAL=true"); - - for (int i = 0; i < 100; i++) { - conn.createStatement().execute("UPSERT INTO " + fullTableName1 + " VALUES (NEXT VALUE FOR " + sequenceName + ", " + (i%10) + ")"); - } - conn.commit(); - conn.setAutoCommit(true); - int upsertCount = conn.createStatement().executeUpdate("UPSERT INTO " + fullTableName2 + " SELECT pk, val FROM " + fullTableName1); - assertEquals(100,upsertCount); - conn.close(); - } - - @Test - public void testTransactionalTableMetadata() throws SQLException { - - try (Connection conn = DriverManager.getConnection(getUrl())) { - String transactTableName = generateUniqueName(); - Statement stmt = conn.createStatement(); - stmt.execute("CREATE TABLE " + transactTableName + " (k VARCHAR PRIMARY KEY, v1 VARCHAR, v2 VARCHAR) " + - "TRANSACTIONAL=true"); - conn.commit(); - - DatabaseMetaData dbmd = conn.getMetaData(); - ResultSet rs = dbmd.getTables(null, null, StringUtil.escapeLike(transactTableName), null); - assertTrue(rs.next()); - assertEquals("Transactional table was not marked as transactional in JDBC API.", - "true", rs.getString(PhoenixDatabaseMetaData.TRANSACTIONAL)); - - String nonTransactTableName = generateUniqueName(); - Statement stmt2 = conn.createStatement(); - stmt2.execute("CREATE TABLE " + nonTransactTableName + "(k VARCHAR PRIMARY KEY, v1 VARCHAR, v2 VARCHAR) "); - conn.commit(); - - ResultSet rs2 = dbmd.getTables(null, null, StringUtil.escapeLike(nonTransactTableName), null); - assertTrue(rs2.next()); - assertEquals("Non-transactional table was marked as transactional in JDBC API.", - "false", rs2.getString(PhoenixDatabaseMetaData.TRANSACTIONAL)); - } - } - - @Test - public void testInflightPartialEval() throws SQLException { - - try (Connection conn = DriverManager.getConnection(getUrl())) { - String transactTableName = generateUniqueName(); - Statement stmt = conn.createStatement(); - stmt.execute("CREATE TABLE " + transactTableName + " (k VARCHAR PRIMARY KEY, v1 VARCHAR, v2 VARCHAR) " + - "TRANSACTIONAL=true"); - - try (Connection conn1 = DriverManager.getConnection(getUrl()); Connection conn2 = DriverManager.getConnection(getUrl())) { - conn1.createStatement().execute("UPSERT INTO " + transactTableName + " VALUES ('a','b','x')"); - // Select to force uncommitted data to be written - ResultSet rs = conn1.createStatement().executeQuery("SELECT * FROM " + transactTableName); - assertTrue(rs.next()); - assertEquals("a", rs.getString(1)); - assertEquals("b", rs.getString(2)); - assertFalse(rs.next()); - - conn2.createStatement().execute("UPSERT INTO " + transactTableName + " VALUES ('a','c','x')"); - // Select to force uncommitted data to be written - rs = conn2.createStatement().executeQuery("SELECT * FROM " + transactTableName ); - assertTrue(rs.next()); - assertEquals("a", rs.getString(1)); - assertEquals("c", rs.getString(2)); - assertFalse(rs.next()); - - // If the AndExpression were to see the uncommitted row from conn2, the filter would - // filter the row out early and no longer continue to evaluate other cells due to - // the way partial evaluation holds state. - rs = conn1.createStatement().executeQuery("SELECT * FROM " + transactTableName + " WHERE v1 != 'c' AND v2 = 'x'"); - assertTrue(rs.next()); - assertEquals("a", rs.getString(1)); - assertEquals("b", rs.getString(2)); - assertFalse(rs.next()); - - // Same as above for conn1 data - rs = conn2.createStatement().executeQuery("SELECT * FROM " + transactTableName + " WHERE v1 != 'b' AND v2 = 'x'"); - assertTrue(rs.next()); - assertEquals("a", rs.getString(1)); - assertEquals("c", rs.getString(2)); - assertFalse(rs.next()); - } - - } - } - - - @Test - public void testOnDupKeyForTransactionalTable() throws Exception { - // TODO: we should support having a transactional table defined for a connectionless connection - try (Connection conn = DriverManager.getConnection(getUrl())) { - String transactTableName = generateUniqueName(); - conn.createStatement().execute("CREATE TABLE " + transactTableName + " (k integer not null primary key, v bigint) TRANSACTIONAL=true"); - conn.createStatement().execute("UPSERT INTO " + transactTableName + " VALUES(0,0) ON DUPLICATE KEY UPDATE v = v + 1"); - fail(); - } catch (SQLException e) { - assertEquals(SQLExceptionCode.CANNOT_USE_ON_DUP_KEY_FOR_TRANSACTIONAL.getErrorCode(), e.getErrorCode()); - } - } - -} +} \ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/it/java/org/apache/phoenix/tx/TxCheckpointIT.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/it/java/org/apache/phoenix/tx/TxCheckpointIT.java b/phoenix-core/src/it/java/org/apache/phoenix/tx/TxCheckpointIT.java index 246ecd4..cb3b4b3 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/tx/TxCheckpointIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/tx/TxCheckpointIT.java @@ -36,6 +36,7 @@ import org.apache.phoenix.end2end.ParallelStatsDisabledIT; import org.apache.phoenix.execute.MutationState; import org.apache.phoenix.jdbc.PhoenixConnection; import org.apache.phoenix.query.QueryServices; +import org.apache.phoenix.schema.PTableImpl; import org.apache.phoenix.util.PropertiesUtil; import org.apache.phoenix.util.SchemaUtil; import org.apache.tephra.Transaction.VisibilityLevel; @@ -48,12 +49,25 @@ import org.junit.runners.Parameterized.Parameters; public class TxCheckpointIT extends ParallelStatsDisabledIT { private final boolean localIndex; - private final boolean mutable; + private final String tableDDLOptions; - public TxCheckpointIT(boolean localIndex, boolean mutable) { + public TxCheckpointIT(boolean localIndex, boolean mutable, boolean columnEncoded) { + StringBuilder optionBuilder = new StringBuilder(); this.localIndex = localIndex; - this.mutable = mutable; - + if (!columnEncoded) { + if (optionBuilder.length()!=0) + optionBuilder.append(","); + optionBuilder.append("COLUMN_ENCODED_BYTES=0"); + } + if (!mutable) { + if (optionBuilder.length()!=0) + optionBuilder.append(","); + optionBuilder.append("IMMUTABLE_ROWS=true"); + if (!columnEncoded) { + optionBuilder.append(",IMMUTABLE_STORAGE_SCHEME="+PTableImpl.ImmutableStorageScheme.ONE_CELL_PER_COLUMN); + } + } + this.tableDDLOptions = optionBuilder.toString(); } private static Connection getConnection() throws SQLException { @@ -66,10 +80,11 @@ public class TxCheckpointIT extends ParallelStatsDisabledIT { return conn; } - @Parameters(name="TxCheckpointIT_localIndex={0},mutable={1}") // name is used by failsafe as file name in reports + @Parameters(name="TxCheckpointIT_localIndex={0},mutable={1},columnEncoded={2}") // name is used by failsafe as file name in reports public static Collection<Boolean[]> data() { return Arrays.asList(new Boolean[][] { - { false, false }, { false, true }, { true, false }, { true, true } + { false, false, false }, { false, false, true }, { false, true, false }, { false, true, true }, + { true, false, false }, { true, false, true }, { true, true, false }, { true, true, true } }); } @@ -86,7 +101,7 @@ public class TxCheckpointIT extends ParallelStatsDisabledIT { Connection conn = getConnection(props); conn.setAutoCommit(true); conn.createStatement().execute("CREATE SEQUENCE "+seqName); - conn.createStatement().execute("CREATE TABLE " + fullTableName + "(pk INTEGER PRIMARY KEY, val INTEGER)"+(!mutable? " IMMUTABLE_ROWS=true" : "")); + conn.createStatement().execute("CREATE TABLE " + fullTableName + "(pk INTEGER PRIMARY KEY, val INTEGER)" + tableDDLOptions); conn.createStatement().execute("CREATE "+(localIndex? "LOCAL " : "")+"INDEX " + indexName + " ON " + fullTableName + "(val)"); conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES (NEXT VALUE FOR " + seqName + ",1)"); @@ -117,12 +132,11 @@ public class TxCheckpointIT extends ParallelStatsDisabledIT { } private void testRollbackOfUncommittedDelete(String indexDDL, String fullTableName) throws Exception { - Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES); Connection conn = getConnection(); conn.setAutoCommit(false); try { Statement stmt = conn.createStatement(); - stmt.execute("CREATE TABLE " + fullTableName + "(k VARCHAR PRIMARY KEY, v1 VARCHAR, v2 VARCHAR)"+(!mutable? " IMMUTABLE_ROWS=true" : "")); + stmt.execute("CREATE TABLE " + fullTableName + "(k VARCHAR PRIMARY KEY, v1 VARCHAR, v2 VARCHAR)" + tableDDLOptions); stmt.execute(indexDDL); stmt.executeUpdate("upsert into " + fullTableName + " values('x1', 'y1', 'a1')"); @@ -206,13 +220,11 @@ public class TxCheckpointIT extends ParallelStatsDisabledIT { String tableName = "TBL_" + generateUniqueName(); String indexName = "IDX_" + generateUniqueName(); String fullTableName = SchemaUtil.getTableName(tableName, tableName); - Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES); try (Connection conn = getConnection()) { conn.setAutoCommit(false); Statement stmt = conn.createStatement(); - stmt.execute("CREATE TABLE " + fullTableName + "(ID BIGINT NOT NULL PRIMARY KEY, v1 VARCHAR, v2 VARCHAR)" - + (!mutable ? " IMMUTABLE_ROWS=true" : "")); + stmt.execute("CREATE TABLE " + fullTableName + "(ID BIGINT NOT NULL PRIMARY KEY, v1 VARCHAR, v2 VARCHAR)" + tableDDLOptions); stmt.execute("CREATE " + (localIndex ? "LOCAL " : "") + "INDEX " + indexName + " ON " + fullTableName + " (v1) INCLUDE(v2)"); @@ -301,10 +313,8 @@ public class TxCheckpointIT extends ParallelStatsDisabledIT { try (Connection conn = getConnection()) { conn.setAutoCommit(false); Statement stmt = conn.createStatement(); - stmt.execute("CREATE TABLE " + fullTableName + "1(ID1 BIGINT NOT NULL PRIMARY KEY, FK1A INTEGER, FK1B INTEGER)" - + (!mutable ? " IMMUTABLE_ROWS=true" : "")); - stmt.execute("CREATE TABLE " + fullTableName + "2(ID2 BIGINT NOT NULL PRIMARY KEY, FK2 INTEGER)" - + (!mutable ? " IMMUTABLE_ROWS=true" : "")); + stmt.execute("CREATE TABLE " + fullTableName + "1(ID1 BIGINT NOT NULL PRIMARY KEY, FK1A INTEGER, FK1B INTEGER)" + tableDDLOptions); + stmt.execute("CREATE TABLE " + fullTableName + "2(ID2 BIGINT NOT NULL PRIMARY KEY, FK2 INTEGER)" + tableDDLOptions); stmt.execute("CREATE " + (localIndex ? "LOCAL " : "") + "INDEX " + indexName + " ON " + fullTableName + "1 (FK1B)"); http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/main/java/org/apache/phoenix/cache/ServerCacheClient.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/cache/ServerCacheClient.java b/phoenix-core/src/main/java/org/apache/phoenix/cache/ServerCacheClient.java index 0383251..18e4034 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/cache/ServerCacheClient.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/cache/ServerCacheClient.java @@ -56,6 +56,7 @@ import org.apache.phoenix.coprocessor.generated.ServerCachingProtos.AddServerCac import org.apache.phoenix.coprocessor.generated.ServerCachingProtos.RemoveServerCacheRequest; import org.apache.phoenix.coprocessor.generated.ServerCachingProtos.RemoveServerCacheResponse; import org.apache.phoenix.coprocessor.generated.ServerCachingProtos.ServerCachingService; +import org.apache.phoenix.index.IndexMaintainer; import org.apache.phoenix.jdbc.PhoenixConnection; import org.apache.phoenix.job.JobManager.JobCallable; import org.apache.phoenix.memory.MemoryManager.MemoryChunk; @@ -215,6 +216,7 @@ public class ServerCacheClient { } builder.setCacheId(ByteStringer.wrap(cacheId)); builder.setCachePtr(org.apache.phoenix.protobuf.ProtobufUtil.toProto(cachePtr)); + builder.setHasProtoBufIndexMaintainer(true); ServerCacheFactoryProtos.ServerCacheFactory.Builder svrCacheFactoryBuider = ServerCacheFactoryProtos.ServerCacheFactory.newBuilder(); svrCacheFactoryBuider.setClassName(cacheFactory.getClass().getName()); builder.setCacheFactory(svrCacheFactoryBuider.build()); http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/main/java/org/apache/phoenix/cache/TenantCache.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/cache/TenantCache.java b/phoenix-core/src/main/java/org/apache/phoenix/cache/TenantCache.java index 5c33967..d30f5dd 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/cache/TenantCache.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/cache/TenantCache.java @@ -36,7 +36,7 @@ import org.apache.phoenix.memory.MemoryManager; public interface TenantCache { MemoryManager getMemoryManager(); Closeable getServerCache(ImmutableBytesPtr cacheId); - Closeable addServerCache(ImmutableBytesPtr cacheId, ImmutableBytesWritable cachePtr, byte[] txState, ServerCacheFactory cacheFactory) throws SQLException; + Closeable addServerCache(ImmutableBytesPtr cacheId, ImmutableBytesWritable cachePtr, byte[] txState, ServerCacheFactory cacheFactory, boolean useProtoForIndexMaintainer) throws SQLException; void removeServerCache(ImmutableBytesPtr cacheId); void removeAllServerCache(); } http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/main/java/org/apache/phoenix/cache/TenantCacheImpl.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/cache/TenantCacheImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/cache/TenantCacheImpl.java index 658b4cc..3d178f6 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/cache/TenantCacheImpl.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/cache/TenantCacheImpl.java @@ -104,11 +104,11 @@ public class TenantCacheImpl implements TenantCache { } @Override - public Closeable addServerCache(ImmutableBytesPtr cacheId, ImmutableBytesWritable cachePtr, byte[] txState, ServerCacheFactory cacheFactory) throws SQLException { + public Closeable addServerCache(ImmutableBytesPtr cacheId, ImmutableBytesWritable cachePtr, byte[] txState, ServerCacheFactory cacheFactory, boolean useProtoForIndexMaintainer) throws SQLException { MemoryChunk chunk = this.getMemoryManager().allocate(cachePtr.getLength() + txState.length); boolean success = false; try { - Closeable element = cacheFactory.newCache(cachePtr, txState, chunk); + Closeable element = cacheFactory.newCache(cachePtr, txState, chunk, useProtoForIndexMaintainer); getServerCaches().put(cacheId, element); success = true; return element; http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/main/java/org/apache/phoenix/compile/CreateTableCompiler.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/CreateTableCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/CreateTableCompiler.java index 07df105..b482998 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/CreateTableCompiler.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/CreateTableCompiler.java @@ -38,6 +38,7 @@ import org.apache.phoenix.expression.IsNullExpression; import org.apache.phoenix.expression.KeyValueColumnExpression; import org.apache.phoenix.expression.LiteralExpression; import org.apache.phoenix.expression.RowKeyColumnExpression; +import org.apache.phoenix.expression.SingleCellColumnExpression; import org.apache.phoenix.expression.visitor.StatelessTraverseNoExpressionVisitor; import org.apache.phoenix.jdbc.PhoenixConnection; import org.apache.phoenix.jdbc.PhoenixStatement; @@ -239,7 +240,7 @@ public class CreateTableCompiler { } } - private static class ViewWhereExpressionVisitor extends StatelessTraverseNoExpressionVisitor<Boolean> { + public static class ViewWhereExpressionVisitor extends StatelessTraverseNoExpressionVisitor<Boolean> { private boolean isUpdatable = true; private final PTable table; private int position; @@ -318,13 +319,18 @@ public class CreateTableCompiler { @Override public Boolean visit(KeyValueColumnExpression node) { try { - this.position = table.getColumnFamily(node.getColumnFamily()).getColumn(node.getColumnName()).getPosition(); + this.position = table.getColumnFamily(node.getColumnFamily()).getPColumnForColumnQualifier(node.getColumnQualifier()).getPosition(); } catch (SQLException e) { throw new RuntimeException(e); // Impossible } return Boolean.TRUE; } + @Override + public Boolean visit(SingleCellColumnExpression node) { + return visit(node.getKeyValueExpression()); + } + } private static class VarbinaryDatum implements PDatum { http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java index 602cd6b..cee545a 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java @@ -585,7 +585,7 @@ public class DeleteCompiler { if (ptr.getLength() > 0) { byte[] uuidValue = ServerCacheClient.generateId(); context.getScan().setAttribute(PhoenixIndexCodec.INDEX_UUID, uuidValue); - context.getScan().setAttribute(PhoenixIndexCodec.INDEX_MD, ptr.get()); + context.getScan().setAttribute(PhoenixIndexCodec.INDEX_PROTO_MD, ptr.get()); context.getScan().setAttribute(BaseScannerRegionObserver.TX_STATE, txState); } ResultIterator iterator = aggPlan.iterator(); http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/main/java/org/apache/phoenix/compile/ExpressionCompiler.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/ExpressionCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/ExpressionCompiler.java index c05918b..fb4c542 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/ExpressionCompiler.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/ExpressionCompiler.java @@ -117,12 +117,26 @@ import org.apache.phoenix.schema.PColumn; import org.apache.phoenix.schema.PDatum; import org.apache.phoenix.schema.PTable; import org.apache.phoenix.schema.PTable.IndexType; +import org.apache.phoenix.schema.PTable.ImmutableStorageScheme; import org.apache.phoenix.schema.PTableType; import org.apache.phoenix.schema.RowKeyValueAccessor; import org.apache.phoenix.schema.SortOrder; import org.apache.phoenix.schema.TableRef; import org.apache.phoenix.schema.TypeMismatchException; -import org.apache.phoenix.schema.types.*; +import org.apache.phoenix.schema.types.PArrayDataType; +import org.apache.phoenix.schema.types.PBoolean; +import org.apache.phoenix.schema.types.PChar; +import org.apache.phoenix.schema.types.PDataType; +import org.apache.phoenix.schema.types.PDate; +import org.apache.phoenix.schema.types.PDecimal; +import org.apache.phoenix.schema.types.PDouble; +import org.apache.phoenix.schema.types.PLong; +import org.apache.phoenix.schema.types.PTimestamp; +import org.apache.phoenix.schema.types.PUnsignedTimestamp; +import org.apache.phoenix.schema.types.PVarbinary; +import org.apache.phoenix.schema.types.PVarchar; +import org.apache.phoenix.schema.types.PhoenixArray; +import org.apache.phoenix.util.EncodedColumnsUtil; import org.apache.phoenix.util.ExpressionUtil; import org.apache.phoenix.util.IndexUtil; import org.apache.phoenix.util.SchemaUtil; @@ -386,7 +400,7 @@ public class ExpressionCompiler extends UnsupportedAllParseNodeVisitor<Expressio } protected void addColumn(PColumn column) { - context.getScan().addColumn(column.getFamilyName().getBytes(), column.getName().getBytes()); + EncodedColumnsUtil.setColumns(column, context.getCurrentTable().getTable(), context.getScan()); } @Override http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/main/java/org/apache/phoenix/compile/FromCompiler.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/FromCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/FromCompiler.java index 8e4d9aa..8ba0e12 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/FromCompiler.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/FromCompiler.java @@ -17,6 +17,8 @@ */ package org.apache.phoenix.compile; +import static org.apache.phoenix.schema.PTable.QualifierEncodingScheme.NON_ENCODED_QUALIFIERS; + import java.sql.SQLException; import java.sql.SQLFeatureNotSupportedException; import java.util.ArrayList; @@ -28,6 +30,7 @@ import java.util.Map; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.client.HTableInterface; +import org.apache.hadoop.hbase.util.Bytes; import org.apache.phoenix.coprocessor.MetaDataProtocol; import org.apache.phoenix.coprocessor.MetaDataProtocol.MetaDataMutationResult; import org.apache.phoenix.expression.Expression; @@ -71,6 +74,8 @@ 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.QualifierEncodingScheme; +import org.apache.phoenix.schema.PTable.ImmutableStorageScheme; import org.apache.phoenix.schema.PTableImpl; import org.apache.phoenix.schema.PTableKey; import org.apache.phoenix.schema.PTableType; @@ -125,10 +130,12 @@ public class FromCompiler { throw new ColumnNotFoundException(schemaName, tableName, null, colName); } + @Override public PFunction resolveFunction(String functionName) throws SQLException { throw new FunctionNotFoundException(functionName); } + @Override public boolean hasUDFs() { return false; } @@ -257,7 +264,7 @@ public class FromCompiler { Expression sourceExpression = projector.getColumnProjector(column.getPosition()).getExpression(); PColumnImpl projectedColumn = new PColumnImpl(column.getName(), column.getFamilyName(), sourceExpression.getDataType(), sourceExpression.getMaxLength(), sourceExpression.getScale(), sourceExpression.isNullable(), - column.getPosition(), sourceExpression.getSortOrder(), column.getArraySize(), column.getViewConstant(), column.isViewReferenced(), column.getExpressionStr(), column.isRowTimestamp(), column.isDynamic()); + column.getPosition(), sourceExpression.getSortOrder(), column.getArraySize(), column.getViewConstant(), column.isViewReferenced(), column.getExpressionStr(), column.isRowTimestamp(), column.isDynamic(), column.getColumnQualifierBytes()); projectedColumns.add(projectedColumn); } PTable t = PTableImpl.makePTable(table, projectedColumns); @@ -332,26 +339,28 @@ public class FromCompiler { private final String alias; private final List<PSchema> schemas; - public SingleTableColumnResolver(PhoenixConnection connection, NamedTableNode table, long timeStamp, Map<String, UDFParseNode> udfParseNodes, boolean isNamespaceMapped) throws SQLException { - super(connection, 0, false, udfParseNodes); - List<PColumnFamily> families = Lists.newArrayListWithExpectedSize(table.getDynamicColumns().size()); - for (ColumnDef def : table.getDynamicColumns()) { - if (def.getColumnDefName().getFamilyName() != null) { - families.add(new PColumnFamilyImpl(PNameFactory.newName(def.getColumnDefName().getFamilyName()),Collections.<PColumn>emptyList())); - } + public SingleTableColumnResolver(PhoenixConnection connection, NamedTableNode table, long timeStamp, Map<String, UDFParseNode> udfParseNodes, boolean isNamespaceMapped) throws SQLException { + super(connection, 0, false, udfParseNodes); + List<PColumnFamily> families = Lists.newArrayListWithExpectedSize(table.getDynamicColumns().size()); + for (ColumnDef def : table.getDynamicColumns()) { + if (def.getColumnDefName().getFamilyName() != null) { + families.add(new PColumnFamilyImpl(PNameFactory.newName(def.getColumnDefName().getFamilyName()),Collections.<PColumn>emptyList()));//, NON_ENCODED_QUALIFIERS)); + } } Long scn = connection.getSCN(); String schema = table.getName().getSchemaName(); if (connection.getSchema() != null) { schema = schema != null ? schema : connection.getSchema(); } - PTable theTable = new PTableImpl(connection.getTenantId(), schema, table.getName().getTableName(), + // Storage scheme and encoding scheme don't matter here since the PTable is being used only for the purposes of create table. + // The actual values of these two will be determined by the metadata client. + PTable theTable = new PTableImpl(connection.getTenantId(), schema, table.getName().getTableName(), scn == null ? HConstants.LATEST_TIMESTAMP : scn, families, isNamespaceMapped); - theTable = this.addDynamicColumns(table.getDynamicColumns(), theTable); - alias = null; - tableRefs = ImmutableList.of(new TableRef(alias, theTable, timeStamp, !table.getDynamicColumns().isEmpty())); - schemas = ImmutableList.of(new PSchema(theTable.getSchemaName().toString(), timeStamp)); - } + theTable = this.addDynamicColumns(table.getDynamicColumns(), theTable); + alias = null; + tableRefs = ImmutableList.of(new TableRef(alias, theTable, timeStamp, !table.getDynamicColumns().isEmpty())); + schemas = ImmutableList.of(new PSchema(theTable.getSchemaName().toString(), timeStamp)); + } public SingleTableColumnResolver(PhoenixConnection connection, NamedTableNode tableNode, boolean updateCacheImmediately) throws SQLException { this(connection, tableNode, updateCacheImmediately, 0, new HashMap<String,UDFParseNode>(1)); @@ -447,8 +456,8 @@ public class FromCompiler { } PColumn column = resolveCF - ? tableRef.getTable().getColumnFamily(tableName).getColumn(colName) - : tableRef.getTable().getColumn(colName); + ? tableRef.getTable().getColumnFamily(tableName).getPColumnForColumnName(colName) + : tableRef.getTable().getPColumnForColumnName(colName); return new ColumnRef(tableRef, column.getPosition()); } @@ -672,7 +681,7 @@ public class FromCompiler { familyName = PNameFactory.newName(family); } allcolumns.add(new PColumnImpl(name, familyName, dynColumn.getDataType(), dynColumn.getMaxLength(), - dynColumn.getScale(), dynColumn.isNull(), position, dynColumn.getSortOrder(), dynColumn.getArraySize(), null, false, dynColumn.getExpression(), false, true)); + dynColumn.getScale(), dynColumn.isNull(), position, dynColumn.getSortOrder(), dynColumn.getArraySize(), null, false, dynColumn.getExpression(), false, true, Bytes.toBytes(dynColumn.getColumnDefName().getColumnName()))); position++; } theTable = PTableImpl.makePTable(theTable, allcolumns); @@ -774,16 +783,17 @@ public class FromCompiler { // referenced by an outer wild-card select. alias = String.valueOf(position); } + PName name = PNameFactory.newName(alias); PColumnImpl column = new PColumnImpl(PNameFactory.newName(alias), PNameFactory.newName(QueryConstants.DEFAULT_COLUMN_FAMILY), - null, 0, 0, true, position++, SortOrder.ASC, null, null, false, null, false, false); + null, 0, 0, true, position++, SortOrder.ASC, null, null, false, null, false, false, name.getBytes()); columns.add(column); } PTable t = PTableImpl.makePTable(null, PName.EMPTY_NAME, PName.EMPTY_NAME, PTableType.SUBQUERY, null, MetaDataProtocol.MIN_TABLE_TIMESTAMP, PTable.INITIAL_SEQ_NUM, null, null, columns, null, null, Collections.<PTable> emptyList(), false, Collections.<PName> emptyList(), null, null, false, false, false, null, null, null, false, false, 0, 0L, SchemaUtil - .isNamespaceMappingEnabled(PTableType.SUBQUERY, connection.getQueryServices().getProps()), null, false); + .isNamespaceMappingEnabled(PTableType.SUBQUERY, connection.getQueryServices().getProps()), null, false, ImmutableStorageScheme.ONE_CELL_PER_COLUMN, QualifierEncodingScheme.NON_ENCODED_QUALIFIERS, PTable.EncodedCQCounter.NULL_COUNTER); String alias = subselectNode.getAlias(); TableRef tableRef = new TableRef(alias, t, MetaDataProtocol.MIN_TABLE_TIMESTAMP, false); @@ -858,7 +868,7 @@ public class FromCompiler { while (iterator.hasNext()) { TableRef tableRef = iterator.next(); try { - PColumn column = tableRef.getTable().getColumn(colName); + PColumn column = tableRef.getTable().getPColumnForColumnName(colName); if (theTableRef != null) { throw new AmbiguousColumnException(colName); } theTableRef = tableRef; theColumnPosition = column.getPosition(); @@ -871,12 +881,12 @@ public class FromCompiler { } else { try { TableRef tableRef = resolveTable(schemaName, tableName); - PColumn column = tableRef.getTable().getColumn(colName); + PColumn column = tableRef.getTable().getPColumnForColumnName(colName); return new ColumnRef(tableRef, column.getPosition()); } catch (TableNotFoundException e) { // Try using the tableName as a columnFamily reference instead ColumnFamilyRef cfRef = resolveColumnFamily(schemaName, tableName); - PColumn column = cfRef.getFamily().getColumn(colName); + PColumn column = cfRef.getFamily().getPColumnForColumnName(colName); return new ColumnRef(cfRef.getTableRef(), column.getPosition()); } } http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java index e8c05ca..eef604b 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java @@ -17,6 +17,9 @@ */ package org.apache.phoenix.compile; +import static org.apache.phoenix.schema.PTable.QualifierEncodingScheme.NON_ENCODED_QUALIFIERS; +import static org.apache.phoenix.schema.PTable.ImmutableStorageScheme.ONE_CELL_PER_COLUMN; + import java.sql.SQLException; import java.util.ArrayList; import java.util.Collections; @@ -76,6 +79,8 @@ 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.QualifierEncodingScheme; +import org.apache.phoenix.schema.PTable.ImmutableStorageScheme; import org.apache.phoenix.schema.PTableImpl; import org.apache.phoenix.schema.PTableType; import org.apache.phoenix.schema.ProjectedColumn; @@ -93,6 +98,7 @@ import org.apache.phoenix.schema.types.PTimestamp; import org.apache.phoenix.schema.types.PTinyint; import org.apache.phoenix.schema.types.PVarbinary; import org.apache.phoenix.schema.types.PVarchar; +import org.apache.phoenix.util.EncodedColumnsUtil; import org.apache.phoenix.util.IndexUtil; import org.apache.phoenix.util.SchemaUtil; @@ -714,7 +720,7 @@ public class JoinCompiler { if (columnRef.getTableRef().equals(tableRef) && !SchemaUtil.isPKColumn(columnRef.getColumn()) && !(columnRef instanceof LocalIndexColumnRef)) { - scan.addColumn(columnRef.getColumn().getFamilyName().getBytes(), columnRef.getColumn().getName().getBytes()); + EncodedColumnsUtil.setColumns(columnRef.getColumn(), tableRef.getTable(), scan); } } } @@ -1284,7 +1290,7 @@ public class JoinCompiler { if (type == JoinType.Full) { for (PColumn c : left.getColumns()) { merged.add(new ProjectedColumn(c.getName(), c.getFamilyName(), - c.getPosition(), true, ((ProjectedColumn) c).getSourceColumnRef())); + c.getPosition(), true, ((ProjectedColumn) c).getSourceColumnRef(), SchemaUtil.isPKColumn(c) ? null : c.getName().getBytes())); } } else { merged.addAll(left.getColumns()); @@ -1294,14 +1300,13 @@ public class JoinCompiler { if (!SchemaUtil.isPKColumn(c)) { PColumn column = new ProjectedColumn(c.getName(), c.getFamilyName(), position++, type == JoinType.Inner ? c.isNullable() : true, - ((ProjectedColumn) c).getSourceColumnRef()); + ((ProjectedColumn) c).getSourceColumnRef(), c.getName().getBytes()); merged.add(column); } } if (left.getBucketNum() != null) { merged.remove(0); } - return PTableImpl.makePTable(left.getTenantId(), left.getSchemaName(), PNameFactory.newName(SchemaUtil.getTableName(left.getName().getString(), right.getName().getString())), left.getType(), left.getIndexState(), left.getTimeStamp(), left.getSequenceNumber(), left.getPKName(), @@ -1310,7 +1315,7 @@ public class JoinCompiler { left.isMultiTenant(), left.getStoreNulls(), left.getViewType(), left.getViewIndexId(), left.getIndexType(), left.rowKeyOrderOptimizable(), left.isTransactional(), left.getUpdateCacheFrequency(), left.getIndexDisableTimestamp(), left.isNamespaceMapped(), - left.getAutoPartitionSeqName(), left.isAppendOnlySchema()); + left.getAutoPartitionSeqName(), left.isAppendOnlySchema(), ONE_CELL_PER_COLUMN, NON_ENCODED_QUALIFIERS, PTable.EncodedCQCounter.NULL_COUNTER); } } http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/main/java/org/apache/phoenix/compile/ListJarsQueryPlan.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/ListJarsQueryPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/ListJarsQueryPlan.java index 2df0671..8265de8 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/ListJarsQueryPlan.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/ListJarsQueryPlan.java @@ -57,6 +57,7 @@ import org.apache.phoenix.query.KeyRange; import org.apache.phoenix.query.QueryServices; 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.RowKeyValueAccessor; import org.apache.phoenix.schema.SortOrder; @@ -77,10 +78,11 @@ public class ListJarsQueryPlan implements QueryPlan { static { List<ExpressionProjector> projectedColumns = new ArrayList<ExpressionProjector>(); + PName colName = PNameFactory.newName("jar_location"); PColumn column = - new PColumnImpl(PNameFactory.newName("jar_location"), null, + new PColumnImpl(colName, null, PVarchar.INSTANCE, null, null, false, 0, SortOrder.getDefault(), 0, null, - false, null, false, false); + false, null, false, false, colName.getBytes()); List<PColumn> columns = new ArrayList<PColumn>(); columns.add(column); Expression expression = http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/main/java/org/apache/phoenix/compile/PostDDLCompiler.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/PostDDLCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/PostDDLCompiler.java index 393499a..e4ef25f 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/PostDDLCompiler.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/PostDDLCompiler.java @@ -50,6 +50,7 @@ import org.apache.phoenix.schema.TableNotFoundException; import org.apache.phoenix.schema.TableRef; import org.apache.phoenix.schema.tuple.Tuple; import org.apache.phoenix.schema.types.PLong; +import org.apache.phoenix.util.EncodedColumnsUtil; import org.apache.phoenix.util.ScanUtil; import org.apache.phoenix.util.TransactionUtil; @@ -176,8 +177,8 @@ public class PostDDLCompiler { @Override public ColumnRef resolveColumn(String schemaName, String tableName, String colName) throws SQLException { PColumn column = tableName != null - ? tableRef.getTable().getColumnFamily(tableName).getColumn(colName) - : tableRef.getTable().getColumn(colName); + ? tableRef.getTable().getColumnFamily(tableName).getPColumnForColumnName(colName) + : tableRef.getTable().getPColumnForColumnName(colName); return new ColumnRef(tableRef, column.getPosition()); } @@ -213,6 +214,7 @@ public class PostDDLCompiler { ScanUtil.setTimeRange(scan, scan.getTimeRange().getMin(), ts); if (emptyCF != null) { scan.setAttribute(BaseScannerRegionObserver.EMPTY_CF, emptyCF); + scan.setAttribute(BaseScannerRegionObserver.EMPTY_COLUMN_QUALIFIER, EncodedColumnsUtil.getEmptyKeyValueInfo(tableRef.getTable()).getFirst()); } ServerCache cache = null; try { @@ -236,11 +238,12 @@ public class PostDDLCompiler { // data empty column family to stay the same, while the index empty column family // changes. PColumn column = deleteList.get(0); + byte[] cq = column.getColumnQualifierBytes(); if (emptyCF == null) { - scan.addColumn(column.getFamilyName().getBytes(), column.getName().getBytes()); + scan.addColumn(column.getFamilyName().getBytes(), cq); } scan.setAttribute(BaseScannerRegionObserver.DELETE_CF, column.getFamilyName().getBytes()); - scan.setAttribute(BaseScannerRegionObserver.DELETE_CQ, column.getName().getBytes()); + scan.setAttribute(BaseScannerRegionObserver.DELETE_CQ, cq); } } List<byte[]> columnFamilies = Lists.newArrayListWithExpectedSize(tableRef.getTable().getColumnFamilies().size()); http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/main/java/org/apache/phoenix/compile/PostLocalIndexDDLCompiler.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/PostLocalIndexDDLCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/PostLocalIndexDDLCompiler.java index 81dbe0d..7e3c3b2 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/PostLocalIndexDDLCompiler.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/PostLocalIndexDDLCompiler.java @@ -31,6 +31,7 @@ import org.apache.phoenix.jdbc.PhoenixConnection; import org.apache.phoenix.jdbc.PhoenixStatement; import org.apache.phoenix.jdbc.PhoenixStatement.Operation; import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.schema.PTable.ImmutableStorageScheme; import org.apache.phoenix.schema.SortOrder; import org.apache.phoenix.schema.TableRef; import org.apache.phoenix.schema.tuple.Tuple; @@ -77,12 +78,16 @@ public class PostLocalIndexDDLCompiler { // rows per region as a result. The value of the attribute will be our persisted // index maintainers. // Define the LOCAL_INDEX_BUILD as a new static in BaseScannerRegionObserver - scan.setAttribute(BaseScannerRegionObserver.LOCAL_INDEX_BUILD, ByteUtil.copyKeyBytesIfNecessary(ptr)); + scan.setAttribute(BaseScannerRegionObserver.LOCAL_INDEX_BUILD_PROTO, ByteUtil.copyKeyBytesIfNecessary(ptr)); // By default, we'd use a FirstKeyOnly filter as nothing else needs to be projected for count(*). // However, in this case, we need to project all of the data columns that contribute to the index. IndexMaintainer indexMaintainer = index.getIndexMaintainer(dataTable, connection); for (ColumnReference columnRef : indexMaintainer.getAllColumns()) { - scan.addColumn(columnRef.getFamily(), columnRef.getQualifier()); + if (index.getImmutableStorageScheme() == ImmutableStorageScheme.SINGLE_CELL_ARRAY_WITH_OFFSETS) { + scan.addFamily(columnRef.getFamily()); + } else { + scan.addColumn(columnRef.getFamily(), columnRef.getQualifier()); + } } // Go through MutationPlan abstraction so that we can create local indexes http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/main/java/org/apache/phoenix/compile/ProjectionCompiler.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/ProjectionCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/ProjectionCompiler.java index 99a9731..eef1ada 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/ProjectionCompiler.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/ProjectionCompiler.java @@ -24,11 +24,9 @@ import java.sql.SQLException; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; -import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.NavigableSet; -import java.util.Set; import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; @@ -44,14 +42,11 @@ import org.apache.phoenix.expression.Expression; import org.apache.phoenix.expression.KeyValueColumnExpression; import org.apache.phoenix.expression.LiteralExpression; import org.apache.phoenix.expression.ProjectedColumnExpression; -import org.apache.phoenix.expression.aggregator.ClientAggregators; -import org.apache.phoenix.expression.aggregator.ServerAggregators; +import org.apache.phoenix.expression.SingleCellColumnExpression; import org.apache.phoenix.expression.function.ArrayIndexFunction; -import org.apache.phoenix.expression.function.SingleAggregateFunction; import org.apache.phoenix.expression.visitor.ExpressionVisitor; import org.apache.phoenix.expression.visitor.ProjectedColumnExpressionVisitor; import org.apache.phoenix.expression.visitor.ReplaceArrayFunctionExpressionVisitor; -import org.apache.phoenix.expression.visitor.SingleAggregateFunctionVisitor; import org.apache.phoenix.jdbc.PhoenixConnection; import org.apache.phoenix.parse.AliasedNode; import org.apache.phoenix.parse.BindParseNode; @@ -78,6 +73,7 @@ import org.apache.phoenix.schema.PColumnFamily; import org.apache.phoenix.schema.PDatum; import org.apache.phoenix.schema.PName; import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.schema.PTable.ImmutableStorageScheme; import org.apache.phoenix.schema.PTable.IndexType; import org.apache.phoenix.schema.PTableKey; import org.apache.phoenix.schema.PTableType; @@ -92,9 +88,7 @@ import org.apache.phoenix.util.IndexUtil; import org.apache.phoenix.util.SchemaUtil; import org.apache.phoenix.util.SizedUtil; -import com.google.common.collect.Iterators; import com.google.common.collect.Lists; -import com.google.common.collect.Sets; /** @@ -217,7 +211,7 @@ public class ProjectionCompiler { PColumn indexColumn = null; ColumnRef ref = null; try { - indexColumn = index.getColumn(indexColName); + indexColumn = index.getPColumnForColumnName(indexColName); ref = new ColumnRef(tableRef, indexColumn.getPosition()); } catch (ColumnNotFoundException e) { if (index.getIndexType() == IndexType.LOCAL) { @@ -289,7 +283,7 @@ public class ProjectionCompiler { ColumnRef ref = null; String indexColumnFamily = null; try { - indexColumn = index.getColumn(indexColName); + indexColumn = index.getPColumnForColumnName(indexColName); ref = new ColumnRef(tableRef, indexColumn.getPosition()); indexColumnFamily = indexColumn.getFamilyName() == null ? null : indexColumn.getFamilyName().getString(); } catch (ColumnNotFoundException e) { @@ -484,11 +478,13 @@ public class ProjectionCompiler { } } else { for (byte[] cq : entry.getValue()) { - PColumn column = family.getColumn(cq); - Integer maxLength = column.getMaxLength(); - int byteSize = column.getDataType().isFixedWidth() ? maxLength == null ? column.getDataType().getByteSize() : maxLength : RowKeySchema.ESTIMATED_VARIABLE_LENGTH_SIZE; - estimatedByteSize += SizedUtil.KEY_VALUE_SIZE + estimatedKeySize + byteSize; - } + //if (!Bytes.equals(cq, ByteUtil.EMPTY_BYTE_ARRAY) || cq.length > 0) { + PColumn column = family.getPColumnForColumnQualifier(cq); + Integer maxLength = column.getMaxLength(); + int byteSize = column.getDataType().isFixedWidth() ? maxLength == null ? column.getDataType().getByteSize() : maxLength : RowKeySchema.ESTIMATED_VARIABLE_LENGTH_SIZE; + estimatedByteSize += SizedUtil.KEY_VALUE_SIZE + estimatedKeySize + byteSize; + } + //} } } boolean isProjectEmptyKeyValue = false; @@ -663,7 +659,14 @@ public class ProjectionCompiler { public Void visit(ProjectedColumnExpression expression) { if (expression.getDataType().isArrayType()) { indexProjectedColumns.add(expression); - KeyValueColumnExpression keyValueColumnExpression = new KeyValueColumnExpression(expression.getColumn()); + PColumn col = expression.getColumn(); + PTable table = context.getCurrentTable().getTable(); + KeyValueColumnExpression keyValueColumnExpression; + if (table.getImmutableStorageScheme() != ImmutableStorageScheme.ONE_CELL_PER_COLUMN) { + keyValueColumnExpression = new SingleCellColumnExpression(col, col.getName().getString(), table.getEncodingScheme()); + } else { + keyValueColumnExpression = new KeyValueColumnExpression(col); + } indexKVs.add(keyValueColumnExpression); copyOfChildren.set(0, keyValueColumnExpression); Integer count = arrayExpressionCounts.get(expression); http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java index 2258f28..5126c8b 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java @@ -94,7 +94,7 @@ public class QueryCompiler { */ private static final String LOAD_COLUMN_FAMILIES_ON_DEMAND_ATTR = "_ondemand_"; private final PhoenixStatement statement; - private final Scan scan; + private final Scan scan; private final Scan originalScan; private final ColumnResolver resolver; private final SelectStatement select; http://git-wip-us.apache.org/repos/asf/phoenix/blob/b49fc0d1/phoenix-core/src/main/java/org/apache/phoenix/compile/TraceQueryPlan.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/TraceQueryPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/TraceQueryPlan.java index ed5cda9..8fb435d 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/TraceQueryPlan.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/TraceQueryPlan.java @@ -51,6 +51,7 @@ import org.apache.phoenix.parse.TraceStatement; import org.apache.phoenix.query.KeyRange; 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.RowKeyValueAccessor; import org.apache.phoenix.schema.SortOrder; @@ -75,10 +76,11 @@ public class TraceQueryPlan implements QueryPlan { private static final RowProjector TRACE_PROJECTOR; static { List<ExpressionProjector> projectedColumns = new ArrayList<ExpressionProjector>(); + PName colName = PNameFactory.newName(MetricInfo.TRACE.columnName); PColumn column = new PColumnImpl(PNameFactory.newName(MetricInfo.TRACE.columnName), null, PLong.INSTANCE, null, null, false, 0, SortOrder.getDefault(), 0, null, - false, null, false, false); + false, null, false, false, colName.getBytes()); List<PColumn> columns = new ArrayList<PColumn>(); columns.add(column); Expression expression =
