| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.tajo.engine.query; |
| |
| import com.google.common.base.Optional; |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.hbase.HColumnDescriptor; |
| import org.apache.hadoop.hbase.HConstants; |
| import org.apache.hadoop.hbase.HTableDescriptor; |
| import org.apache.hadoop.hbase.MiniHBaseCluster; |
| import org.apache.hadoop.hbase.TableName; |
| import org.apache.hadoop.hbase.client.*; |
| import org.apache.hadoop.hbase.filter.Filter; |
| import org.apache.hadoop.hbase.filter.InclusiveStopFilter; |
| import org.apache.hadoop.hbase.master.HMaster; |
| import org.apache.hadoop.hbase.regionserver.HRegionServer; |
| import org.apache.tajo.IntegrationTest; |
| import org.apache.tajo.QueryTestCaseBase; |
| import org.apache.tajo.TajoTestingCluster; |
| import org.apache.tajo.catalog.Schema; |
| import org.apache.tajo.catalog.TableDesc; |
| import org.apache.tajo.common.TajoDataTypes.Type; |
| import org.apache.tajo.conf.TajoConf; |
| import org.apache.tajo.datum.TextDatum; |
| import org.apache.tajo.error.Errors.ResultCode; |
| import org.apache.tajo.exception.TajoException; |
| import org.apache.tajo.exception.UnavailableTableLocationException; |
| import org.apache.tajo.plan.expr.*; |
| import org.apache.tajo.plan.logical.ScanNode; |
| import org.apache.tajo.storage.StorageConstants; |
| import org.apache.tajo.storage.Tablespace; |
| import org.apache.tajo.storage.TablespaceManager; |
| import org.apache.tajo.storage.fragment.Fragment; |
| import org.apache.tajo.storage.hbase.*; |
| import org.apache.tajo.util.Bytes; |
| import org.apache.tajo.util.KeyValueSet; |
| import org.junit.AfterClass; |
| import org.junit.BeforeClass; |
| import org.junit.Test; |
| import org.junit.experimental.categories.Category; |
| |
| import java.io.IOException; |
| import java.net.InetAddress; |
| import java.net.URI; |
| import java.sql.ResultSet; |
| import java.text.DecimalFormat; |
| import java.util.*; |
| |
| import static org.junit.Assert.*; |
| |
| @Category(IntegrationTest.class) |
| public class TestHBaseTable extends QueryTestCaseBase { |
| private static final Log LOG = LogFactory.getLog(TestHBaseTable.class); |
| |
| private static String tableSpaceUri; |
| private static String hostName,zkPort; |
| |
| @BeforeClass |
| public static void beforeClass() throws IOException { |
| try { |
| testingCluster.getHBaseUtil().startHBaseCluster(); |
| hostName = InetAddress.getLocalHost().getHostName(); |
| zkPort = testingCluster.getHBaseUtil().getConf().get(HConstants.ZOOKEEPER_CLIENT_PORT); |
| assertNotNull(hostName); |
| assertNotNull(zkPort); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| } |
| |
| tableSpaceUri = "hbase:zk://" + hostName + ":" + zkPort; |
| HBaseTablespace hBaseTablespace = new HBaseTablespace("cluster1", URI.create(tableSpaceUri), null); |
| hBaseTablespace.init(new TajoConf(testingCluster.getHBaseUtil().getConf())); |
| TablespaceManager.addTableSpaceForTest(hBaseTablespace); |
| } |
| |
| @AfterClass |
| public static void afterClass() { |
| try { |
| testingCluster.getHBaseUtil().stopHBaseCluster(); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| } |
| } |
| |
| @Test |
| public void testVerifyCreateHBaseTableRequiredMeta() throws Exception { |
| try { |
| executeString("CREATE TABLE hbase_mapped_table1 (col1 text, col2 text) TABLESPACE cluster1 USING hbase").close(); |
| fail("hbase table must have 'table' meta"); |
| } catch (TajoException e) { |
| assertEquals(e.getErrorCode(), ResultCode.MISSING_TABLE_PROPERTY); |
| } |
| |
| try { |
| executeString("CREATE TABLE hbase_mapped_table1 (col1 text, col2 text) TABLESPACE cluster1 " + |
| "USING hbase " + |
| "WITH ('table'='hbase_table')").close(); |
| |
| fail("hbase table must have 'columns' meta"); |
| } catch (TajoException e) { |
| assertEquals(e.getErrorCode(), ResultCode.MISSING_TABLE_PROPERTY); |
| } |
| } |
| |
| @Test |
| public void testCreateHBaseTable() throws Exception { |
| executeString( |
| "CREATE TABLE hbase_mapped_table1 (col1 text, col2 text, col3 text, col4 text) TABLESPACE cluster1 " + |
| "USING hbase WITH ('table'='hbase_table', 'columns'=':key,col2:a,col3:,col2:b')").close(); |
| |
| assertTableExists("hbase_mapped_table1"); |
| |
| HTableDescriptor hTableDesc = testingCluster.getHBaseUtil().getTableDescriptor("hbase_table"); |
| assertNotNull(hTableDesc); |
| assertEquals("hbase_table", hTableDesc.getNameAsString()); |
| |
| HColumnDescriptor[] hColumns = hTableDesc.getColumnFamilies(); |
| // col1 is mapped to rowkey |
| assertEquals(2, hColumns.length); |
| assertEquals("col2", hColumns[0].getNameAsString()); |
| assertEquals("col3", hColumns[1].getNameAsString()); |
| |
| executeString("DROP TABLE hbase_mapped_table1 PURGE").close(); |
| |
| HBaseAdmin hAdmin = new HBaseAdmin(testingCluster.getHBaseUtil().getConf()); |
| try { |
| assertFalse(hAdmin.tableExists("hbase_table")); |
| } finally { |
| hAdmin.close(); |
| } |
| } |
| |
| @Test |
| public void testCreateNotExistsExternalHBaseTable() throws Exception { |
| Optional<Tablespace> existing = TablespaceManager.removeTablespaceForTest("cluster1"); |
| assertTrue(existing.isPresent()); |
| |
| try { |
| String sql = String.format( |
| "CREATE EXTERNAL TABLE external_hbase_mapped_table1 (col1 text, col2 text, col3 text, col4 text) " + |
| "USING hbase WITH ('table'='external_hbase_table', 'columns'=':key,col2:a,col3:,col2:b') " + |
| "LOCATION '%s/external_hbase_table'", tableSpaceUri); |
| try { |
| executeString(sql).close(); |
| fail("External table should be a existed table."); |
| } catch (Throwable e) { |
| assertTrue(e instanceof UnavailableTableLocationException); |
| } |
| } finally { |
| TablespaceManager.addTableSpaceForTest(existing.get()); |
| } |
| } |
| |
| @Test |
| public void testCreateRowFieldWithNonText() throws Exception { |
| try { |
| executeString("CREATE TABLE hbase_mapped_table2 (rk1 int4, rk2 text, col3 text, col4 text) " + |
| "TABLESPACE cluster1 USING hbase WITH ('table'='hbase_table', 'columns'='0:key#b,1:key,col3:,col2:b', " + |
| "'hbase.rowkey.delimiter'='_')").close(); |
| fail("Key field type should be TEXT type"); |
| } catch (Exception e) { |
| assertTrue(e.getMessage().indexOf("Key field type should be TEXT type") >= 0); |
| } |
| } |
| |
| @Test |
| public void testCreateExternalHBaseTable() throws Exception { |
| Optional<Tablespace> existing = TablespaceManager.removeTablespaceForTest("cluster1"); |
| assertTrue(existing.isPresent()); |
| |
| try { |
| HTableDescriptor hTableDesc = new HTableDescriptor(TableName.valueOf("external_hbase_table_not_purge")); |
| hTableDesc.addFamily(new HColumnDescriptor("col1")); |
| hTableDesc.addFamily(new HColumnDescriptor("col2")); |
| hTableDesc.addFamily(new HColumnDescriptor("col3")); |
| testingCluster.getHBaseUtil().createTable(hTableDesc); |
| |
| String sql = String.format( |
| "CREATE EXTERNAL TABLE external_hbase_mapped_table (rk text, col1 text, col2 text, col3 text) " + |
| "USING hbase WITH ('table'='external_hbase_table_not_purge', 'columns'=':key,col1:a,col2:,col3:b') " + |
| "LOCATION '%s/external_hbase_table'", tableSpaceUri); |
| executeString(sql).close(); |
| |
| assertTableExists("external_hbase_mapped_table"); |
| |
| executeString("DROP TABLE external_hbase_mapped_table").close(); |
| |
| HBaseAdmin hAdmin = new HBaseAdmin(testingCluster.getHBaseUtil().getConf()); |
| try { |
| assertTrue(hAdmin.tableExists("external_hbase_table_not_purge")); |
| hAdmin.disableTable("external_hbase_table_not_purge"); |
| hAdmin.deleteTable("external_hbase_table_not_purge"); |
| } finally { |
| hAdmin.close(); |
| } |
| |
| } finally { |
| TablespaceManager.addTableSpaceForTest(existing.get()); |
| } |
| } |
| |
| private void putData(HTableInterface htable, int rownum) throws IOException { |
| for (int i = 0; i < rownum; i++) { |
| Put put = new Put(String.valueOf(i).getBytes()); |
| put.add("col1".getBytes(), "a".getBytes(), ("a-" + i).getBytes()); |
| put.add("col1".getBytes(), "b".getBytes(), ("b-" + i).getBytes()); |
| put.add("col2".getBytes(), "k1".getBytes(), ("k1-" + i).getBytes()); |
| put.add("col2".getBytes(), "k2".getBytes(), ("k2-" + i).getBytes()); |
| put.add("col3".getBytes(), "b".getBytes(), ("b-" + i).getBytes()); |
| htable.put(put); |
| } |
| } |
| |
| @Test |
| public void testGetTableVolume() throws Exception { |
| final String tableName = "external_hbase_table"; |
| |
| Optional<Tablespace> existing = TablespaceManager.removeTablespaceForTest("cluster1"); |
| assertTrue(existing.isPresent()); |
| |
| try { |
| HTableDescriptor hTableDesc = new HTableDescriptor(TableName.valueOf(tableName)); |
| hTableDesc.addFamily(new HColumnDescriptor("col1")); |
| hTableDesc.addFamily(new HColumnDescriptor("col2")); |
| hTableDesc.addFamily(new HColumnDescriptor("col3")); |
| testingCluster.getHBaseUtil().createTable(hTableDesc); |
| |
| String sql = String.format( |
| "CREATE EXTERNAL TABLE external_hbase_mapped_table (rk text, col1 text, col2 text, col3 text) " + |
| "USING hbase WITH ('table'='external_hbase_table', 'columns'=':key,col1:a,col2:,col3:b') " + |
| "LOCATION '%s/external_hbase_table'", tableSpaceUri); |
| executeString(sql).close(); |
| |
| assertTableExists("external_hbase_mapped_table"); |
| |
| HBaseTablespace tablespace = (HBaseTablespace)existing.get(); |
| HConnection hconn = ((HBaseTablespace)existing.get()).getConnection(); |
| |
| try (HTableInterface htable = hconn.getTable(tableName)) { |
| htable.setAutoFlushTo(true); |
| putData(htable, 4000); |
| } |
| hconn.close(); |
| |
| Thread.sleep(3000); // sleep here for up-to-date region server load. It may not be a problem in real cluster. |
| |
| TableDesc createdTable = client.getTableDesc("external_hbase_mapped_table"); |
| assertNotNull(tablespace); |
| long volume = tablespace.getTableVolume(createdTable, Optional.<EvalNode>absent()); |
| assertTrue(volume > 0 || volume == -1); |
| executeString("DROP TABLE external_hbase_mapped_table PURGE").close(); |
| |
| } finally { |
| TablespaceManager.addTableSpaceForTest(existing.get()); |
| } |
| } |
| |
| @Test |
| public void testSimpleSelectQuery() throws Exception { |
| Optional<Tablespace> existing = TablespaceManager.removeTablespaceForTest("cluster1"); |
| assertTrue(existing.isPresent()); |
| |
| try { |
| HTableDescriptor hTableDesc = new HTableDescriptor(TableName.valueOf("external_hbase_table")); |
| hTableDesc.addFamily(new HColumnDescriptor("col1")); |
| hTableDesc.addFamily(new HColumnDescriptor("col2")); |
| hTableDesc.addFamily(new HColumnDescriptor("col3")); |
| testingCluster.getHBaseUtil().createTable(hTableDesc); |
| |
| String sql = String.format( |
| "CREATE EXTERNAL TABLE external_hbase_mapped_table (rk text, col1 text, col2 text, col3 text) " + |
| "USING hbase WITH ('table'='external_hbase_table', 'columns'=':key,col1:a,col2:,col3:b') " + |
| "LOCATION '%s/external_hbase_table'", tableSpaceUri); |
| executeString(sql).close(); |
| |
| assertTableExists("external_hbase_mapped_table"); |
| |
| HConnection hconn = ((HBaseTablespace)existing.get()).getConnection(); |
| HTableInterface htable = hconn.getTable("external_hbase_table"); |
| |
| try { |
| putData(htable, 100); |
| ResultSet res = executeString("select * from external_hbase_mapped_table where rk > '20'"); |
| assertResultSet(res); |
| cleanupQuery(res); |
| } finally { |
| executeString("DROP TABLE external_hbase_mapped_table PURGE").close(); |
| htable.close(); |
| hconn.close(); |
| } |
| } finally { |
| TablespaceManager.addTableSpaceForTest(existing.get()); |
| } |
| } |
| |
| @Test |
| public void testBinaryMappedQuery() throws Exception { |
| Optional<Tablespace> existing = TablespaceManager.removeTablespaceForTest("cluster1"); |
| assertTrue(existing.isPresent()); |
| |
| try { |
| HTableDescriptor hTableDesc = new HTableDescriptor(TableName.valueOf("external_hbase_table")); |
| hTableDesc.addFamily(new HColumnDescriptor("col1")); |
| hTableDesc.addFamily(new HColumnDescriptor("col2")); |
| hTableDesc.addFamily(new HColumnDescriptor("col3")); |
| testingCluster.getHBaseUtil().createTable(hTableDesc); |
| |
| String sql = String.format( |
| "CREATE EXTERNAL TABLE external_hbase_mapped_table (rk int8, col1 text, col2 text, col3 int4)\n " + |
| "USING hbase WITH ('table'='external_hbase_table', 'columns'=':key#b,col1:a,col2:,col3:b#b') " + |
| "LOCATION '%s/external_hbase_table'", tableSpaceUri); |
| executeString(sql).close(); |
| |
| assertTableExists("external_hbase_mapped_table"); |
| |
| HConnection hconn = ((HBaseTablespace)existing.get()).getConnection(); |
| HTableInterface htable = hconn.getTable("external_hbase_table"); |
| |
| try { |
| for (int i = 0; i < 100; i++) { |
| Put put = new Put(Bytes.toBytes((long) i)); |
| put.add("col1".getBytes(), "a".getBytes(), ("a-" + i).getBytes()); |
| put.add("col1".getBytes(), "b".getBytes(), ("b-" + i).getBytes()); |
| put.add("col2".getBytes(), "k1".getBytes(), ("k1-" + i).getBytes()); |
| put.add("col2".getBytes(), "k2".getBytes(), ("k2-" + i).getBytes()); |
| put.add("col3".getBytes(), "b".getBytes(), Bytes.toBytes(i)); |
| htable.put(put); |
| } |
| |
| ResultSet res = executeString("select * from external_hbase_mapped_table where rk > 20"); |
| assertResultSet(res); |
| res.close(); |
| |
| //Projection |
| res = executeString("select col3, col2, rk from external_hbase_mapped_table where rk > 95"); |
| |
| String expected = "col3,col2,rk\n" + |
| "-------------------------------\n" + |
| "96,{\"k1\":\"k1-96\", \"k2\":\"k2-96\"},96\n" + |
| "97,{\"k1\":\"k1-97\", \"k2\":\"k2-97\"},97\n" + |
| "98,{\"k1\":\"k1-98\", \"k2\":\"k2-98\"},98\n" + |
| "99,{\"k1\":\"k1-99\", \"k2\":\"k2-99\"},99\n"; |
| |
| assertEquals(expected, resultSetToString(res)); |
| res.close(); |
| |
| } finally { |
| executeString("DROP TABLE external_hbase_mapped_table PURGE").close(); |
| htable.close(); |
| } |
| } finally { |
| TablespaceManager.addTableSpaceForTest(existing.get()); |
| } |
| } |
| |
| @Test |
| public void testColumnKeyValueSelectQuery() throws Exception { |
| Optional<Tablespace> existing = TablespaceManager.removeTablespaceForTest("cluster1"); |
| assertTrue(existing.isPresent()); |
| |
| try { |
| HTableDescriptor hTableDesc = new HTableDescriptor(TableName.valueOf("external_hbase_table")); |
| hTableDesc.addFamily(new HColumnDescriptor("col2")); |
| hTableDesc.addFamily(new HColumnDescriptor("col3")); |
| testingCluster.getHBaseUtil().createTable(hTableDesc); |
| |
| String sql = String.format( |
| "CREATE EXTERNAL TABLE external_hbase_mapped_table (rk1 text, col2_key text, col2_value text, col3 text) " + |
| "USING hbase WITH ('table'='external_hbase_table', 'columns'=':key,col2:key:,col2:value:,col3:', " + |
| "'hbase.rowkey.delimiter'='_') LOCATION '%s/external_hbase_table'", tableSpaceUri); |
| executeString(sql).close(); |
| |
| assertTableExists("external_hbase_mapped_table"); |
| |
| HConnection hconn = ((HBaseTablespace)existing.get()).getConnection(); |
| HTableInterface htable = hconn.getTable("external_hbase_table"); |
| |
| try { |
| for (int i = 0; i < 10; i++) { |
| Put put = new Put(Bytes.toBytes("rk-" + i)); |
| for (int j = 0; j < 5; j++) { |
| put.add("col2".getBytes(), ("key-" + j).getBytes(), Bytes.toBytes("value-" + j)); |
| } |
| put.add("col3".getBytes(), "".getBytes(), ("col3-value-" + i).getBytes()); |
| htable.put(put); |
| } |
| |
| ResultSet res = executeString("select * from external_hbase_mapped_table where rk1 >= 'rk-0'"); |
| assertResultSet(res); |
| cleanupQuery(res); |
| } finally { |
| executeString("DROP TABLE external_hbase_mapped_table PURGE").close(); |
| htable.close(); |
| } |
| } finally { |
| TablespaceManager.addTableSpaceForTest(existing.get()); |
| } |
| } |
| |
| @Test |
| public void testRowFieldSelectQuery() throws Exception { |
| Optional<Tablespace> existing = TablespaceManager.removeTablespaceForTest("cluster1"); |
| assertTrue(existing.isPresent()); |
| |
| try { |
| HTableDescriptor hTableDesc = new HTableDescriptor(TableName.valueOf("external_hbase_table")); |
| hTableDesc.addFamily(new HColumnDescriptor("col3")); |
| testingCluster.getHBaseUtil().createTable(hTableDesc); |
| |
| String sql = String.format( |
| "CREATE EXTERNAL TABLE external_hbase_mapped_table (rk1 text, rk2 text, col3 text) " + |
| "USING hbase WITH ('table'='external_hbase_table', 'columns'='0:key,1:key,col3:a', " + |
| "'hbase.rowkey.delimiter'='_') LOCATION '%s/external_hbase_table'", tableSpaceUri); |
| executeString(sql).close(); |
| |
| assertTableExists("external_hbase_mapped_table"); |
| |
| |
| HConnection hconn = ((HBaseTablespace)existing.get()).getConnection(); |
| HTableInterface htable = hconn.getTable("external_hbase_table"); |
| |
| try { |
| for (int i = 0; i < 100; i++) { |
| Put put = new Put(("field1-" + i + "_field2-" + i).getBytes()); |
| put.add("col3".getBytes(), "a".getBytes(), ("a-" + i).getBytes()); |
| htable.put(put); |
| } |
| |
| ResultSet res = executeString("select * from external_hbase_mapped_table where rk1 > 'field1-20'"); |
| assertResultSet(res); |
| cleanupQuery(res); |
| } finally { |
| executeString("DROP TABLE external_hbase_mapped_table PURGE").close(); |
| htable.close(); |
| } |
| } finally { |
| TablespaceManager.addTableSpaceForTest(existing.get()); |
| } |
| } |
| |
| @Test |
| public void testIndexPredication() throws Exception { |
| String sql = |
| "CREATE TABLE hbase_mapped_table (rk text, col1 text, col2 text, col3 text) " + |
| "TABLESPACE cluster1 USING hbase WITH ('table'='hbase_table', 'columns'=':key,col1:a,col2:,col3:b', " + |
| "'hbase.split.rowkeys'='010,040,060,080') "; |
| executeString(sql).close(); |
| |
| |
| assertTableExists("hbase_mapped_table"); |
| HBaseAdmin hAdmin = new HBaseAdmin(testingCluster.getHBaseUtil().getConf()); |
| hAdmin.tableExists("hbase_table"); |
| |
| HTable htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table"); |
| try { |
| org.apache.hadoop.hbase.util.Pair<byte[][], byte[][]> keys = htable.getStartEndKeys(); |
| assertEquals(5, keys.getFirst().length); |
| |
| DecimalFormat df = new DecimalFormat("000"); |
| for (int i = 0; i < 100; i++) { |
| Put put = new Put(String.valueOf(df.format(i)).getBytes()); |
| put.add("col1".getBytes(), "a".getBytes(), ("a-" + i).getBytes()); |
| put.add("col1".getBytes(), "b".getBytes(), ("b-" + i).getBytes()); |
| put.add("col2".getBytes(), "k1".getBytes(), ("k1-" + i).getBytes()); |
| put.add("col2".getBytes(), "k2".getBytes(), ("k2-" + i).getBytes()); |
| put.add("col3".getBytes(), "b".getBytes(), ("b-" + i).getBytes()); |
| htable.put(put); |
| } |
| assertIndexPredication(false); |
| |
| ResultSet res = executeString("select * from hbase_mapped_table where rk >= '020' and rk <= '055'"); |
| assertResultSet(res); |
| res.close(); |
| |
| res = executeString("select * from hbase_mapped_table where rk = '021'"); |
| String expected = "rk,col1,col2,col3\n" + |
| "-------------------------------\n" + |
| "021,a-21,{\"k1\":\"k1-21\", \"k2\":\"k2-21\"},b-21\n"; |
| |
| assertEquals(expected, resultSetToString(res)); |
| res.close(); |
| } finally { |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| htable.close(); |
| hAdmin.close(); |
| } |
| } |
| |
| @Test |
| public void testCompositeRowIndexPredication() throws Exception { |
| |
| executeString("CREATE TABLE hbase_mapped_table (rk text, rk2 text, col1 text, col2 text, col3 text) " + |
| "TABLESPACE cluster1 USING hbase WITH ('table'='hbase_table', 'columns'='0:key,1:key,col1:a,col2:,col3:b', " + |
| "'hbase.split.rowkeys'='010,040,060,080', " + |
| "'hbase.rowkey.delimiter'='_')").close(); |
| |
| assertTableExists("hbase_mapped_table"); |
| HBaseAdmin hAdmin = new HBaseAdmin(testingCluster.getHBaseUtil().getConf()); |
| hAdmin.tableExists("hbase_table"); |
| |
| HTable htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table"); |
| try { |
| org.apache.hadoop.hbase.util.Pair<byte[][], byte[][]> keys = htable.getStartEndKeys(); |
| assertEquals(5, keys.getFirst().length); |
| |
| DecimalFormat df = new DecimalFormat("000"); |
| for (int i = 0; i < 100; i++) { |
| Put put = new Put((df.format(i) + "_" + df.format(i)).getBytes()); |
| put.add("col1".getBytes(), "a".getBytes(), ("a-" + i).getBytes()); |
| put.add("col1".getBytes(), "b".getBytes(), ("b-" + i).getBytes()); |
| put.add("col2".getBytes(), "k1".getBytes(), ("k1-" + i).getBytes()); |
| put.add("col2".getBytes(), "k2".getBytes(), ("k2-" + i).getBytes()); |
| put.add("col3".getBytes(), "b".getBytes(), ("b-" + i).getBytes()); |
| htable.put(put); |
| } |
| |
| Scan scan = new Scan(); |
| scan.setStartRow("021".getBytes()); |
| scan.setStopRow(("021_" + new String(new char[]{Character.MAX_VALUE})).getBytes()); |
| Filter filter = new InclusiveStopFilter(scan.getStopRow()); |
| scan.setFilter(filter); |
| |
| ResultScanner scanner = htable.getScanner(scan); |
| Result result = scanner.next(); |
| assertNotNull(result); |
| assertEquals("021_021", new String(result.getRow())); |
| scanner.close(); |
| |
| assertIndexPredication(true); |
| |
| ResultSet res = executeString("select * from hbase_mapped_table where rk = '021'"); |
| String expected = "rk,rk2,col1,col2,col3\n" + |
| "-------------------------------\n" + |
| "021,021,a-21,{\"k1\":\"k1-21\", \"k2\":\"k2-21\"},b-21\n"; |
| |
| assertEquals(expected, resultSetToString(res)); |
| res.close(); |
| } finally { |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| htable.close(); |
| hAdmin.close(); |
| } |
| } |
| |
| private void assertIndexPredication(boolean isCompositeRowKey) throws Exception { |
| String postFix = isCompositeRowKey ? "_" + new String(new char[]{Character.MAX_VALUE}) : ""; |
| TableDesc tableDesc = catalog.getTableDesc(getCurrentDatabase(), "hbase_mapped_table"); |
| |
| ScanNode scanNode = new ScanNode(1); |
| |
| // where rk = '021' |
| EvalNode evalNodeEq = new BinaryEval(EvalType.EQUAL, new FieldEval(tableDesc.getLogicalSchema().getColumn("rk")), |
| new ConstEval(new TextDatum("021"))); |
| scanNode.setQual(evalNodeEq); |
| Tablespace tablespace = TablespaceManager.getByName("cluster1"); |
| List<Fragment> fragments = tablespace.getSplits("hbase_mapped_table", tableDesc, scanNode.getQual()); |
| assertEquals(1, fragments.size()); |
| assertEquals("021", new String(((HBaseFragment)fragments.get(0)).getStartRow())); |
| assertEquals("021" + postFix, new String(((HBaseFragment)fragments.get(0)).getStopRow())); |
| |
| // where rk >= '020' and rk <= '055' |
| EvalNode evalNode1 = new BinaryEval(EvalType.GEQ, new FieldEval(tableDesc.getLogicalSchema().getColumn("rk")), |
| new ConstEval(new TextDatum("020"))); |
| EvalNode evalNode2 = new BinaryEval(EvalType.LEQ, new FieldEval(tableDesc.getLogicalSchema().getColumn("rk")), |
| new ConstEval(new TextDatum("055"))); |
| EvalNode evalNodeA = new BinaryEval(EvalType.AND, evalNode1, evalNode2); |
| scanNode.setQual(evalNodeA); |
| |
| fragments = tablespace.getSplits("hbase_mapped_table", tableDesc, scanNode.getQual()); |
| assertEquals(2, fragments.size()); |
| HBaseFragment fragment1 = (HBaseFragment) fragments.get(0); |
| assertEquals("020", new String(fragment1.getStartRow())); |
| assertEquals("040", new String(fragment1.getStopRow())); |
| |
| HBaseFragment fragment2 = (HBaseFragment) fragments.get(1); |
| assertEquals("040", new String(fragment2.getStartRow())); |
| assertEquals("055" + postFix, new String(fragment2.getStopRow())); |
| |
| // where (rk >= '020' and rk <= '055') or rk = '075' |
| EvalNode evalNode3 = new BinaryEval(EvalType.EQUAL, new FieldEval(tableDesc.getLogicalSchema().getColumn("rk")), |
| new ConstEval(new TextDatum("075"))); |
| EvalNode evalNodeB = new BinaryEval(EvalType.OR, evalNodeA, evalNode3); |
| scanNode.setQual(evalNodeB); |
| fragments = tablespace.getSplits("hbase_mapped_table", tableDesc, scanNode.getQual()); |
| assertEquals(3, fragments.size()); |
| fragment1 = (HBaseFragment) fragments.get(0); |
| assertEquals("020", new String(fragment1.getStartRow())); |
| assertEquals("040", new String(fragment1.getStopRow())); |
| |
| fragment2 = (HBaseFragment) fragments.get(1); |
| assertEquals("040", new String(fragment2.getStartRow())); |
| assertEquals("055" + postFix, new String(fragment2.getStopRow())); |
| |
| HBaseFragment fragment3 = (HBaseFragment) fragments.get(2); |
| assertEquals("075", new String(fragment3.getStartRow())); |
| assertEquals("075" + postFix, new String(fragment3.getStopRow())); |
| |
| |
| // where (rk >= '020' and rk <= '055') or (rk >= '072' and rk <= '078') |
| EvalNode evalNode4 = new BinaryEval(EvalType.GEQ, new FieldEval(tableDesc.getLogicalSchema().getColumn("rk")), |
| new ConstEval(new TextDatum("072"))); |
| EvalNode evalNode5 = new BinaryEval(EvalType.LEQ, new FieldEval(tableDesc.getLogicalSchema().getColumn("rk")), |
| new ConstEval(new TextDatum("078"))); |
| EvalNode evalNodeC = new BinaryEval(EvalType.AND, evalNode4, evalNode5); |
| EvalNode evalNodeD = new BinaryEval(EvalType.OR, evalNodeA, evalNodeC); |
| scanNode.setQual(evalNodeD); |
| fragments = tablespace.getSplits("hbase_mapped_table", tableDesc, scanNode.getQual()); |
| assertEquals(3, fragments.size()); |
| |
| fragment1 = (HBaseFragment) fragments.get(0); |
| assertEquals("020", new String(fragment1.getStartRow())); |
| assertEquals("040", new String(fragment1.getStopRow())); |
| |
| fragment2 = (HBaseFragment) fragments.get(1); |
| assertEquals("040", new String(fragment2.getStartRow())); |
| assertEquals("055" + postFix, new String(fragment2.getStopRow())); |
| |
| fragment3 = (HBaseFragment) fragments.get(2); |
| assertEquals("072", new String(fragment3.getStartRow())); |
| assertEquals("078" + postFix, new String(fragment3.getStopRow())); |
| |
| // where (rk >= '020' and rk <= '055') or (rk >= '057' and rk <= '059') |
| evalNode4 = new BinaryEval(EvalType.GEQ, new FieldEval(tableDesc.getLogicalSchema().getColumn("rk")), |
| new ConstEval(new TextDatum("057"))); |
| evalNode5 = new BinaryEval(EvalType.LEQ, new FieldEval(tableDesc.getLogicalSchema().getColumn("rk")), |
| new ConstEval(new TextDatum("059"))); |
| evalNodeC = new BinaryEval(EvalType.AND, evalNode4, evalNode5); |
| evalNodeD = new BinaryEval(EvalType.OR, evalNodeA, evalNodeC); |
| scanNode.setQual(evalNodeD); |
| fragments = tablespace.getSplits("hbase_mapped_table", tableDesc, scanNode.getQual()); |
| assertEquals(2, fragments.size()); |
| |
| fragment1 = (HBaseFragment) fragments.get(0); |
| assertEquals("020", new String(fragment1.getStartRow())); |
| assertEquals("040", new String(fragment1.getStopRow())); |
| |
| fragment2 = (HBaseFragment) fragments.get(1); |
| assertEquals("040", new String(fragment2.getStartRow())); |
| assertEquals("059" + postFix, new String(fragment2.getStopRow())); |
| } |
| |
| @Test |
| public void testNonForwardQuery() throws Exception { |
| executeString("CREATE TABLE hbase_mapped_table1 (rk text, col1 text, col2 text, col3 int) " + |
| "TABLESPACE cluster1 USING hbase WITH ('table'='hbase_table1', 'columns'=':key,col1:a,col2:,col3:#b', " + |
| "'hbase.split.rowkeys'='010,040,060,080')").close(); |
| |
| assertTableExists("hbase_mapped_table1"); |
| HBaseAdmin hAdmin = new HBaseAdmin(testingCluster.getHBaseUtil().getConf()); |
| HTable htable = null; |
| try { |
| hAdmin.tableExists("hbase_table1"); |
| htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table1"); |
| org.apache.hadoop.hbase.util.Pair<byte[][], byte[][]> keys = htable.getStartEndKeys(); |
| assertEquals(5, keys.getFirst().length); |
| |
| DecimalFormat df = new DecimalFormat("000"); |
| for (int i = 0; i < 100; i++) { |
| Put put = new Put(String.valueOf(df.format(i)).getBytes()); |
| put.add("col1".getBytes(), "a".getBytes(), ("a-" + i).getBytes()); |
| put.add("col1".getBytes(), "b".getBytes(), ("b-" + i).getBytes()); |
| put.add("col2".getBytes(), "k1".getBytes(), ("k1-" + i).getBytes()); |
| put.add("col2".getBytes(), "k2".getBytes(), ("k2-" + i).getBytes()); |
| put.add("col3".getBytes(), "".getBytes(), Bytes.toBytes(i)); |
| htable.put(put); |
| } |
| |
| ResultSet res = executeString("select * from hbase_mapped_table1"); |
| assertResultSet(res); |
| res.close(); |
| } finally { |
| executeString("DROP TABLE hbase_mapped_table1 PURGE").close(); |
| hAdmin.close(); |
| if (htable == null) { |
| htable.close(); |
| } |
| } |
| } |
| |
| @Test |
| public void testJoin() throws Exception { |
| executeString("CREATE TABLE hbase_mapped_table (rk text, col1 text, col2 text, col3 int8) " + |
| "TABLESPACE cluster1 USING hbase WITH ('table'='hbase_table', 'columns'=':key,col1:a,col2:,col3:b#b', " + |
| "'hbase.split.rowkeys'='010,040,060,080')").close(); |
| |
| assertTableExists("hbase_mapped_table"); |
| HBaseAdmin hAdmin = new HBaseAdmin(testingCluster.getHBaseUtil().getConf()); |
| HTable htable = null; |
| try { |
| hAdmin.tableExists("hbase_table"); |
| htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table"); |
| org.apache.hadoop.hbase.util.Pair<byte[][], byte[][]> keys = htable.getStartEndKeys(); |
| assertEquals(5, keys.getFirst().length); |
| |
| DecimalFormat df = new DecimalFormat("000"); |
| for (int i = 0; i < 100; i++) { |
| Put put = new Put(String.valueOf(df.format(i)).getBytes()); |
| put.add("col1".getBytes(), "a".getBytes(), ("a-" + i).getBytes()); |
| put.add("col1".getBytes(), "b".getBytes(), ("b-" + i).getBytes()); |
| put.add("col2".getBytes(), "k1".getBytes(), ("k1-" + i).getBytes()); |
| put.add("col2".getBytes(), "k2".getBytes(), ("k2-" + i).getBytes()); |
| put.add("col3".getBytes(), "b".getBytes(), Bytes.toBytes((long) i)); |
| htable.put(put); |
| } |
| |
| ResultSet res = executeString("select a.rk, a.col1, a.col2, a.col3, b.l_orderkey, b.l_linestatus " + |
| "from hbase_mapped_table a " + |
| "join default.lineitem b on a.col3 = b.l_orderkey order by a.rk, a.col1, a.col2, a.col3"); |
| assertResultSet(res); |
| res.close(); |
| } finally { |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| hAdmin.close(); |
| if (htable != null) { |
| htable.close(); |
| } |
| } |
| } |
| |
| @Test |
| public void testInsertInto() throws Exception { |
| executeString("CREATE TABLE hbase_mapped_table (rk text, col1 text, col2 text, col3 int4) " + |
| "TABLESPACE cluster1 USING hbase WITH ('table'='hbase_table', 'columns'=':key,col1:a,col2:,col3:b#b')").close(); |
| |
| assertTableExists("hbase_mapped_table"); |
| TableDesc tableDesc = catalog.getTableDesc(getCurrentDatabase(), "hbase_mapped_table"); |
| |
| executeString("insert into hbase_mapped_table " + |
| "select l_orderkey::text, l_shipdate, l_returnflag, l_suppkey from default.lineitem ").close(); |
| |
| HTable htable = null; |
| ResultScanner scanner = null; |
| try { |
| htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table"); |
| |
| Scan scan = new Scan(); |
| scan.addFamily(Bytes.toBytes("col1")); |
| scan.addFamily(Bytes.toBytes("col2")); |
| scan.addFamily(Bytes.toBytes("col3")); |
| scanner = htable.getScanner(scan); |
| |
| assertStrings(resultSetToString(scanner, |
| new byte[][]{null, Bytes.toBytes("col1"), Bytes.toBytes("col2"), Bytes.toBytes("col3")}, |
| new byte[][]{null, Bytes.toBytes("a"), null, Bytes.toBytes("b")}, |
| new boolean[]{false, false, false, true}, tableDesc.getSchema())); |
| |
| } finally { |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| |
| if (scanner != null) { |
| scanner.close(); |
| } |
| |
| if (htable != null) { |
| htable.close(); |
| } |
| } |
| } |
| |
| @Test |
| public void testInsertValues1() throws Exception { |
| executeString("CREATE TABLE hbase_mapped_table (rk text, col1 text, col2 text, col3 int4) " + |
| "TABLESPACE cluster1 USING hbase WITH ('table'='hbase_table', 'columns'=':key,col1:a,col2:,col3:b#b')").close(); |
| |
| assertTableExists("hbase_mapped_table"); |
| TableDesc tableDesc = catalog.getTableDesc(getCurrentDatabase(), "hbase_mapped_table"); |
| |
| executeString("insert into hbase_mapped_table select 'aaa', 'a12', 'a34', 1").close(); |
| executeString("insert into hbase_mapped_table select 'bbb', 'b12', 'b34', 2").close(); |
| executeString("insert into hbase_mapped_table select 'ccc', 'c12', 'c34', 3").close(); |
| executeString("insert into hbase_mapped_table select 'ddd', 'd12', 'd34', 4").close(); |
| |
| HTable htable = null; |
| ResultScanner scanner = null; |
| try { |
| htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table"); |
| |
| Scan scan = new Scan(); |
| scan.addFamily(Bytes.toBytes("col1")); |
| scan.addFamily(Bytes.toBytes("col2")); |
| scan.addFamily(Bytes.toBytes("col3")); |
| scanner = htable.getScanner(scan); |
| |
| assertStrings(resultSetToString(scanner, |
| new byte[][]{null, Bytes.toBytes("col1"), Bytes.toBytes("col2"), Bytes.toBytes("col3")}, |
| new byte[][]{null, Bytes.toBytes("a"), null, Bytes.toBytes("b")}, |
| new boolean[]{false, false, false, true}, tableDesc.getSchema())); |
| |
| } finally { |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| |
| if (scanner != null) { |
| scanner.close(); |
| } |
| |
| if (htable != null) { |
| htable.close(); |
| } |
| } |
| } |
| |
| @Test |
| public void testInsertIntoMultiRegion() throws Exception { |
| executeString("CREATE TABLE hbase_mapped_table (rk text, col1 text) TABLESPACE cluster1 " + |
| "USING hbase WITH ('table'='hbase_table', 'columns'=':key,col1:a', " + |
| "'hbase.split.rowkeys'='010,040,060,080')").close(); |
| |
| assertTableExists("hbase_mapped_table"); |
| TableDesc tableDesc = catalog.getTableDesc(getCurrentDatabase(), "hbase_mapped_table"); |
| |
| // create test table |
| KeyValueSet tableOptions = new KeyValueSet(); |
| tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER); |
| tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N"); |
| |
| Schema schema = new Schema(); |
| schema.addColumn("id", Type.TEXT); |
| schema.addColumn("name", Type.TEXT); |
| List<String> datas = new ArrayList<>(); |
| DecimalFormat df = new DecimalFormat("000"); |
| for (int i = 99; i >= 0; i--) { |
| datas.add(df.format(i) + "|value" + i); |
| } |
| TajoTestingCluster.createTable(getCurrentDatabase() + ".base_table", |
| schema, tableOptions, datas.toArray(new String[datas.size()]), 2); |
| |
| executeString("insert into hbase_mapped_table " + |
| "select id, name from base_table ").close(); |
| |
| HTable htable = null; |
| ResultScanner scanner = null; |
| try { |
| htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table"); |
| |
| Scan scan = new Scan(); |
| scan.addFamily(Bytes.toBytes("col1")); |
| scanner = htable.getScanner(scan); |
| |
| assertStrings(resultSetToString(scanner, |
| new byte[][]{null, Bytes.toBytes("col1")}, |
| new byte[][]{null, Bytes.toBytes("a")}, |
| new boolean[]{false, false}, tableDesc.getSchema())); |
| |
| } finally { |
| executeString("DROP TABLE base_table PURGE").close(); |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| |
| if (scanner != null) { |
| scanner.close(); |
| } |
| |
| if (htable != null) { |
| htable.close(); |
| } |
| } |
| } |
| |
| @Test |
| public void testInsertIntoMultiRegion2() throws Exception { |
| executeString("CREATE TABLE hbase_mapped_table (rk text, col1 text) TABLESPACE cluster1 " + |
| "USING hbase WITH ('table'='hbase_table', 'columns'=':key,col1:a', " + |
| "'hbase.split.rowkeys'='1,2,3,4,5,6,7,8,9')").close(); |
| |
| assertTableExists("hbase_mapped_table"); |
| TableDesc tableDesc = catalog.getTableDesc(getCurrentDatabase(), "hbase_mapped_table"); |
| |
| // create test table |
| KeyValueSet tableOptions = new KeyValueSet(); |
| tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER); |
| tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N"); |
| |
| Schema schema = new Schema(); |
| schema.addColumn("id", Type.TEXT); |
| schema.addColumn("name", Type.TEXT); |
| List<String> datas = new ArrayList<>(); |
| for (int i = 99; i >= 0; i--) { |
| datas.add(i + "|value" + i); |
| } |
| TajoTestingCluster.createTable(getCurrentDatabase() + ".base_table", |
| schema, tableOptions, datas.toArray(new String[datas.size()]), 2); |
| |
| executeString("insert into hbase_mapped_table " + |
| "select id, name from base_table ").close(); |
| |
| HTable htable = null; |
| ResultScanner scanner = null; |
| try { |
| htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table"); |
| |
| Scan scan = new Scan(); |
| scan.addFamily(Bytes.toBytes("col1")); |
| scanner = htable.getScanner(scan); |
| |
| assertStrings(resultSetToString(scanner, |
| new byte[][]{null, Bytes.toBytes("col1")}, |
| new byte[][]{null, Bytes.toBytes("a")}, |
| new boolean[]{false, false}, tableDesc.getSchema())); |
| |
| } finally { |
| executeString("DROP TABLE base_table PURGE").close(); |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| |
| if (scanner != null) { |
| scanner.close(); |
| } |
| |
| if (htable != null) { |
| htable.close(); |
| } |
| } |
| } |
| |
| @Test |
| public void testInsertIntoMultiRegionWithSplitFile() throws Exception { |
| String splitFilePath = currentDatasetPath + "/splits.data"; |
| |
| executeString( |
| "CREATE TABLE hbase_mapped_table (rk text, col1 text) TABLESPACE cluster1 " + |
| "USING hbase WITH ('table'='hbase_table', 'columns'=':key,col1:a', " + |
| "'hbase.split.rowkeys.file'='" + splitFilePath + "')").close(); |
| |
| assertTableExists("hbase_mapped_table"); |
| TableDesc tableDesc = catalog.getTableDesc(getCurrentDatabase(), "hbase_mapped_table"); |
| |
| // create test table |
| KeyValueSet tableOptions = new KeyValueSet(); |
| tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER); |
| tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N"); |
| |
| Schema schema = new Schema(); |
| schema.addColumn("id", Type.TEXT); |
| schema.addColumn("name", Type.TEXT); |
| List<String> datas = new ArrayList<>(); |
| DecimalFormat df = new DecimalFormat("000"); |
| for (int i = 99; i >= 0; i--) { |
| datas.add(df.format(i) + "|value" + i); |
| } |
| TajoTestingCluster.createTable(getCurrentDatabase() + ".base_table", |
| schema, tableOptions, datas.toArray(new String[datas.size()]), 2); |
| |
| executeString("insert into hbase_mapped_table " + |
| "select id, name from base_table ").close(); |
| |
| HTable htable = null; |
| ResultScanner scanner = null; |
| try { |
| htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table"); |
| |
| Scan scan = new Scan(); |
| scan.addFamily(Bytes.toBytes("col1")); |
| scanner = htable.getScanner(scan); |
| |
| assertStrings(resultSetToString(scanner, |
| new byte[][]{null, Bytes.toBytes("col1")}, |
| new byte[][]{null, Bytes.toBytes("a")}, |
| new boolean[]{false, false}, tableDesc.getSchema())); |
| |
| } finally { |
| executeString("DROP TABLE base_table PURGE").close(); |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| |
| if (scanner != null) { |
| scanner.close(); |
| } |
| |
| if (htable != null) { |
| htable.close(); |
| } |
| } |
| } |
| |
| @Test |
| public void testInsertIntoMultiRegionMultiRowFields() throws Exception { |
| executeString( |
| "CREATE TABLE hbase_mapped_table (rk1 text, rk2 text, col1 text) TABLESPACE cluster1 " + |
| "USING hbase WITH ('table'='hbase_table', 'columns'='0:key,1:key,col1:a', " + |
| "'hbase.split.rowkeys'='001,002,003,004,005,006,007,008,009', " + |
| "'hbase.rowkey.delimiter'='_')").close(); |
| |
| assertTableExists("hbase_mapped_table"); |
| TableDesc tableDesc = catalog.getTableDesc(getCurrentDatabase(), "hbase_mapped_table"); |
| |
| // create test table |
| KeyValueSet tableOptions = new KeyValueSet(); |
| tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER); |
| tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N"); |
| |
| Schema schema = new Schema(); |
| schema.addColumn("id1", Type.TEXT); |
| schema.addColumn("id2", Type.TEXT); |
| schema.addColumn("name", Type.TEXT); |
| DecimalFormat df = new DecimalFormat("000"); |
| List<String> datas = new ArrayList<>(); |
| for (int i = 99; i >= 0; i--) { |
| datas.add(df.format(i) + "|" + (i + 100) + "|value" + i); |
| } |
| TajoTestingCluster.createTable(getCurrentDatabase() + ".base_table", |
| schema, tableOptions, datas.toArray(new String[datas.size()]), 2); |
| |
| executeString("insert into hbase_mapped_table " + |
| "select id1, id2, name from base_table ").close(); |
| |
| HTable htable = null; |
| ResultScanner scanner = null; |
| try { |
| htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table"); |
| |
| Scan scan = new Scan(); |
| scan.addFamily(Bytes.toBytes("col1")); |
| scanner = htable.getScanner(scan); |
| |
| assertStrings(resultSetToString(scanner, |
| new byte[][]{null, null, Bytes.toBytes("col1")}, |
| new byte[][]{null, null, Bytes.toBytes("a")}, |
| new boolean[]{false, false, false}, tableDesc.getSchema())); |
| |
| } finally { |
| executeString("DROP TABLE base_table PURGE").close(); |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| |
| if (scanner != null) { |
| scanner.close(); |
| } |
| |
| if (htable != null) { |
| htable.close(); |
| } |
| } |
| } |
| |
| @Test |
| public void testInsertIntoBinaryMultiRegion() throws Exception { |
| executeString("CREATE TABLE hbase_mapped_table (rk int4, col1 text) TABLESPACE cluster1 " + |
| "USING hbase WITH ('table'='hbase_table', 'columns'=':key#b,col1:a', " + |
| "'hbase.split.rowkeys'='1,2,3,4,5,6,7,8,9')").close(); |
| |
| assertTableExists("hbase_mapped_table"); |
| TableDesc tableDesc = catalog.getTableDesc(getCurrentDatabase(), "hbase_mapped_table"); |
| |
| // create test table |
| KeyValueSet tableOptions = new KeyValueSet(); |
| tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER); |
| tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N"); |
| |
| Schema schema = new Schema(); |
| schema.addColumn("id", Type.INT4); |
| schema.addColumn("name", Type.TEXT); |
| List<String> datas = new ArrayList<>(); |
| for (int i = 99; i >= 0; i--) { |
| datas.add(i + "|value" + i); |
| } |
| TajoTestingCluster.createTable(getCurrentDatabase() + ".base_table", |
| schema, tableOptions, datas.toArray(new String[datas.size()]), 2); |
| |
| executeString("insert into hbase_mapped_table " + |
| "select id, name from base_table ").close(); |
| |
| HTable htable = null; |
| ResultScanner scanner = null; |
| try { |
| htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table"); |
| |
| Scan scan = new Scan(); |
| scan.addFamily(Bytes.toBytes("col1")); |
| scanner = htable.getScanner(scan); |
| |
| assertStrings(resultSetToString(scanner, |
| new byte[][]{null, Bytes.toBytes("col1")}, |
| new byte[][]{null, Bytes.toBytes("a")}, |
| new boolean[]{true, false}, tableDesc.getSchema())); |
| |
| } finally { |
| executeString("DROP TABLE base_table PURGE").close(); |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| |
| if (scanner != null) { |
| scanner.close(); |
| } |
| |
| if (htable != null) { |
| htable.close(); |
| } |
| } |
| } |
| |
| @Test |
| public void testInsertIntoColumnKeyValue() throws Exception { |
| executeString( |
| "CREATE TABLE hbase_mapped_table (rk text, col2_key text, col2_value text, col3 text) TABLESPACE cluster1 " + |
| "USING hbase WITH ('table'='hbase_table', 'columns'=':key,col2:key:,col2:value:,col3:', " + |
| "'hbase.rowkey.delimiter'='_')").close(); |
| |
| assertTableExists("hbase_mapped_table"); |
| TableDesc tableDesc = catalog.getTableDesc(getCurrentDatabase(), "hbase_mapped_table"); |
| |
| // create test table |
| KeyValueSet tableOptions = new KeyValueSet(); |
| tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER); |
| tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N"); |
| |
| Schema schema = new Schema(); |
| schema.addColumn("rk", Type.TEXT); |
| schema.addColumn("col2_key", Type.TEXT); |
| schema.addColumn("col2_value", Type.TEXT); |
| schema.addColumn("col3", Type.TEXT); |
| List<String> datas = new ArrayList<>(); |
| for (int i = 20; i >= 0; i--) { |
| for (int j = 0; j < 3; j++) { |
| datas.add(i + "|ck-" + j + "|value-" + j + "|col3-" + i); |
| } |
| } |
| TajoTestingCluster.createTable(getCurrentDatabase() + ".base_table", |
| schema, tableOptions, datas.toArray(new String[datas.size()]), 2); |
| |
| executeString("insert into hbase_mapped_table " + |
| "select rk, col2_key, col2_value, col3 from base_table ").close(); |
| |
| HTable htable = null; |
| ResultScanner scanner = null; |
| try { |
| htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table"); |
| |
| Scan scan = new Scan(); |
| scan.addFamily(Bytes.toBytes("col2")); |
| scan.addFamily(Bytes.toBytes("col3")); |
| scanner = htable.getScanner(scan); |
| |
| assertStrings(resultSetToString(scanner, |
| new byte[][]{null, Bytes.toBytes("col2"), Bytes.toBytes("col3")}, |
| new byte[][]{null, null, null}, |
| new boolean[]{false, false, false}, tableDesc.getSchema())); |
| |
| ResultSet res = executeString("select * from hbase_mapped_table"); |
| |
| String expected = "rk,col2_key,col2_value,col3\n" + |
| "-------------------------------\n" + |
| "0,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-0\n" + |
| "1,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-1\n" + |
| "10,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-10\n" + |
| "11,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-11\n" + |
| "12,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-12\n" + |
| "13,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-13\n" + |
| "14,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-14\n" + |
| "15,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-15\n" + |
| "16,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-16\n" + |
| "17,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-17\n" + |
| "18,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-18\n" + |
| "19,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-19\n" + |
| "2,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-2\n" + |
| "20,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-20\n" + |
| "3,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-3\n" + |
| "4,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-4\n" + |
| "5,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-5\n" + |
| "6,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-6\n" + |
| "7,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-7\n" + |
| "8,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-8\n" + |
| "9,[\"ck-0\", \"ck-1\", \"ck-2\"],[\"value-0\", \"value-1\", \"value-2\"],col3-9\n"; |
| |
| assertEquals(expected, resultSetToString(res)); |
| res.close(); |
| |
| } finally { |
| executeString("DROP TABLE base_table PURGE").close(); |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| |
| if (scanner != null) { |
| scanner.close(); |
| } |
| |
| if (htable != null) { |
| htable.close(); |
| } |
| } |
| } |
| |
| @Test |
| public void testInsertIntoDifferentType() throws Exception { |
| executeString( |
| "CREATE TABLE hbase_mapped_table (rk text, col1 text) TABLESPACE cluster1 " + |
| "USING hbase WITH ('table'='hbase_table', 'columns'=':key,col1:a', " + |
| "'hbase.split.rowkeys'='1,2,3,4,5,6,7,8,9')").close(); |
| |
| assertTableExists("hbase_mapped_table"); |
| |
| // create test table |
| KeyValueSet tableOptions = new KeyValueSet(); |
| tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER); |
| tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N"); |
| |
| Schema schema = new Schema(); |
| schema.addColumn("id", Type.INT4); |
| schema.addColumn("name", Type.TEXT); |
| List<String> datas = new ArrayList<>(); |
| for (int i = 99; i >= 0; i--) { |
| datas.add(i + "|value" + i); |
| } |
| TajoTestingCluster.createTable(getCurrentDatabase() + ".base_table", |
| schema, tableOptions, datas.toArray(new String[datas.size()]), 2); |
| |
| try { |
| executeString("insert into hbase_mapped_table " + |
| "select id, name from base_table ").close(); |
| fail("If inserting data type different with target table data type, should throw exception"); |
| } catch (TajoException e) { |
| assertEquals(ResultCode.DATATYPE_MISMATCH, e.getErrorCode()); |
| } finally { |
| executeString("DROP TABLE base_table PURGE").close(); |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| } |
| } |
| |
| @Test |
| public void testInsertIntoRowField() throws Exception { |
| executeString( |
| "CREATE TABLE hbase_mapped_table (rk1 text, rk2 text, col1 text, col2 text, col3 text) TABLESPACE cluster1 " + |
| "USING hbase WITH ('table'='hbase_table', 'columns'='0:key,1:key,col1:a,col2:,col3:b', " + |
| "'hbase.rowkey.delimiter'='_')").close(); |
| |
| |
| assertTableExists("hbase_mapped_table"); |
| TableDesc tableDesc = catalog.getTableDesc(getCurrentDatabase(), "hbase_mapped_table"); |
| |
| executeString("insert into hbase_mapped_table " + |
| "select l_orderkey::text, l_partkey::text, l_shipdate, l_returnflag, l_suppkey::text from default.lineitem "); |
| |
| HTable htable = null; |
| ResultScanner scanner = null; |
| try { |
| htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table"); |
| |
| Scan scan = new Scan(); |
| scan.addFamily(Bytes.toBytes("col1")); |
| scan.addFamily(Bytes.toBytes("col2")); |
| scan.addFamily(Bytes.toBytes("col3")); |
| scanner = htable.getScanner(scan); |
| |
| assertStrings(resultSetToString(scanner, |
| new byte[][]{null, Bytes.toBytes("col1"), Bytes.toBytes("col2"), Bytes.toBytes("col3")}, |
| new byte[][]{null, Bytes.toBytes("a"), Bytes.toBytes(""), Bytes.toBytes("b")}, |
| new boolean[]{false, false, false, false}, tableDesc.getSchema())); |
| |
| } finally { |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| |
| if (scanner != null) { |
| scanner.close(); |
| } |
| |
| if (htable != null) { |
| htable.close(); |
| } |
| } |
| } |
| |
| @Test |
| public void testCTAS() throws Exception { |
| // create test table |
| KeyValueSet tableOptions = new KeyValueSet(); |
| tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER); |
| tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N"); |
| |
| Schema schema = new Schema(); |
| schema.addColumn("id", Type.TEXT); |
| schema.addColumn("name", Type.TEXT); |
| List<String> datas = new ArrayList<>(); |
| DecimalFormat df = new DecimalFormat("000"); |
| for (int i = 99; i >= 0; i--) { |
| datas.add(df.format(i) + "|value" + i); |
| } |
| TajoTestingCluster.createTable(getCurrentDatabase() + ".base_table", |
| schema, tableOptions, datas.toArray(new String[datas.size()]), 2); |
| |
| executeString( |
| "CREATE TABLE hbase_mapped_table (rk text, col1 text) TABLESPACE cluster1 " + |
| "USING hbase WITH ('table'='hbase_table', 'columns'=':key,col1:a', " + |
| "'hbase.split.rowkeys'='010,040,060,080') as" + |
| " select id, name from base_table" |
| ).close(); |
| |
| assertTableExists("hbase_mapped_table"); |
| TableDesc tableDesc = catalog.getTableDesc(getCurrentDatabase(), "hbase_mapped_table"); |
| |
| HTable htable = null; |
| ResultScanner scanner = null; |
| try { |
| htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table"); |
| |
| Scan scan = new Scan(); |
| scan.addFamily(Bytes.toBytes("col1")); |
| scanner = htable.getScanner(scan); |
| |
| assertStrings(resultSetToString(scanner, |
| new byte[][]{null, Bytes.toBytes("col1")}, |
| new byte[][]{null, Bytes.toBytes("a")}, |
| new boolean[]{false, false}, tableDesc.getSchema())); |
| |
| } finally { |
| executeString("DROP TABLE base_table PURGE").close(); |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| |
| if (scanner != null) { |
| scanner.close(); |
| } |
| |
| if (htable != null) { |
| htable.close(); |
| } |
| |
| // TODO - rollback should support its corresponding hbase table |
| HBaseAdmin hAdmin = new HBaseAdmin(testingCluster.getHBaseUtil().getConf()); |
| if (hAdmin.tableExists("hbase_table")) { |
| hAdmin.disableTable("hbase_table"); |
| hAdmin.deleteTable("hbase_table"); |
| } |
| } |
| } |
| |
| @Test |
| public void testInsertIntoUsingPut() throws Exception { |
| executeString( |
| "CREATE TABLE hbase_mapped_table (rk text, col1 text, col2 text, col3 int4) TABLESPACE cluster1 " + |
| "USING hbase WITH ('table'='hbase_table', 'columns'=':key,col1:a,col2:,col3:b#b')").close(); |
| |
| assertTableExists("hbase_mapped_table"); |
| TableDesc tableDesc = catalog.getTableDesc(getCurrentDatabase(), "hbase_mapped_table"); |
| |
| Map<String, String> sessions = new HashMap<>(); |
| sessions.put(HBaseStorageConstants.INSERT_PUT_MODE, "true"); |
| client.updateSessionVariables(sessions); |
| |
| HTable htable = null; |
| ResultScanner scanner = null; |
| try { |
| executeString( |
| "insert into hbase_mapped_table " + |
| "select l_orderkey::text, l_shipdate, l_returnflag, l_suppkey from default.lineitem" |
| ).close(); |
| |
| htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table"); |
| |
| Scan scan = new Scan(); |
| scan.addFamily(Bytes.toBytes("col1")); |
| scan.addFamily(Bytes.toBytes("col2")); |
| scan.addFamily(Bytes.toBytes("col3")); |
| scanner = htable.getScanner(scan); |
| |
| // result is dirrerent with testInsertInto because l_orderkey is not unique. |
| assertStrings(resultSetToString(scanner, |
| new byte[][]{null, Bytes.toBytes("col1"), Bytes.toBytes("col2"), Bytes.toBytes("col3")}, |
| new byte[][]{null, Bytes.toBytes("a"), null, Bytes.toBytes("b")}, |
| new boolean[]{false, false, false, true}, tableDesc.getSchema())); |
| |
| } finally { |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| |
| client.unsetSessionVariables(Arrays.asList(HBaseStorageConstants.INSERT_PUT_MODE)); |
| |
| if (scanner != null) { |
| scanner.close(); |
| } |
| |
| if (htable != null) { |
| htable.close(); |
| } |
| } |
| } |
| |
| @Test |
| public void testInsertIntoLocation() throws Exception { |
| executeString( |
| "CREATE TABLE hbase_mapped_table (rk text, col1 text, col2 text) TABLESPACE cluster1 " + |
| "USING hbase WITH ('table'='hbase_table', 'columns'=':key,col1:a,col2:', " + |
| "'hbase.split.rowkeys'='010,040,060,080')").close(); |
| |
| assertTableExists("hbase_mapped_table"); |
| |
| try { |
| // create test table |
| KeyValueSet tableOptions = new KeyValueSet(); |
| tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER); |
| tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N"); |
| |
| Schema schema = new Schema(); |
| schema.addColumn("id", Type.TEXT); |
| schema.addColumn("name", Type.TEXT); |
| schema.addColumn("comment", Type.TEXT); |
| List<String> datas = new ArrayList<>(); |
| DecimalFormat df = new DecimalFormat("000"); |
| for (int i = 99; i >= 0; i--) { |
| datas.add(df.format(i) + "|value" + i + "|comment-" + i); |
| } |
| TajoTestingCluster.createTable(getCurrentDatabase() + ".base_table", |
| schema, tableOptions, datas.toArray(new String[datas.size()]), 2); |
| |
| executeString("insert into location '/tmp/hfile_test' " + |
| "select id, name, comment from base_table ").close(); |
| |
| FileSystem fs = testingCluster.getDefaultFileSystem(); |
| Path path = new Path("/tmp/hfile_test"); |
| assertTrue(fs.exists(path)); |
| |
| FileStatus[] files = fs.listStatus(path); |
| assertNotNull(files); |
| assertEquals(2, files.length); |
| |
| int index = 0; |
| for (FileStatus eachFile: files) { |
| assertEquals("/tmp/hfile_test/part-01-00000" + index + "-00" + index, eachFile.getPath().toUri().getPath()); |
| for (FileStatus subFile: fs.listStatus(eachFile.getPath())) { |
| assertTrue(subFile.isFile()); |
| assertTrue(subFile.getLen() > 0); |
| } |
| index++; |
| } |
| } finally { |
| executeString("DROP TABLE base_table PURGE").close(); |
| executeString("DROP TABLE hbase_mapped_table PURGE").close(); |
| } |
| } |
| |
| @Test |
| public void testGetSplitsWhenRestartHBase() throws Exception { |
| executeString("CREATE TABLE hbase_mapped_table1 (rk text, col1 text, col2 text, col3 int) " |
| + "TABLESPACE cluster1 USING hbase WITH ('table'='hbase_table1', 'columns'=':key,col1:a,col2:,col3:#b', " |
| + "'hbase.split.rowkeys'='010,020,030,040,050,060,070,080,090')").close(); |
| |
| assertTableExists("hbase_mapped_table1"); |
| HBaseAdmin hAdmin = new HBaseAdmin(testingCluster.getHBaseUtil().getConf()); |
| HTable htable = null; |
| try { |
| hAdmin.tableExists("hbase_table1"); |
| htable = new HTable(testingCluster.getHBaseUtil().getConf(), "hbase_table1"); |
| org.apache.hadoop.hbase.util.Pair<byte[][], byte[][]> keys = htable.getStartEndKeys(); |
| assertEquals(10, keys.getFirst().length); |
| |
| DecimalFormat df = new DecimalFormat("000"); |
| for (int i = 0; i < 100; i++) { |
| Put put = new Put(String.valueOf(df.format(i)).getBytes()); |
| put.add("col1".getBytes(), "a".getBytes(), ("a-" + i).getBytes()); |
| put.add("col1".getBytes(), "b".getBytes(), ("b-" + i).getBytes()); |
| put.add("col2".getBytes(), "k1".getBytes(), ("k1-" + i).getBytes()); |
| put.add("col2".getBytes(), "k2".getBytes(), ("k2-" + i).getBytes()); |
| put.add("col3".getBytes(), "".getBytes(), Bytes.toBytes(i)); |
| htable.put(put); |
| } |
| |
| ResultSet res = executeString("select * from hbase_mapped_table1"); |
| assertResultSet(res); |
| res.close(); |
| |
| MiniHBaseCluster cluster = testingCluster.getHBaseUtil().getMiniHBaseCluster(); |
| HMaster master = cluster.getMaster(); |
| master.balanceSwitch(true); |
| assertEquals(1, cluster.getLiveRegionServerThreads().size()); |
| HRegionServer orgRegionServer = cluster.getLiveRegionServerThreads().get(0).getRegionServer(); |
| cluster.startRegionServer().waitForServerOnline(); |
| cluster.startRegionServer().waitForServerOnline(); |
| cluster.startRegionServer().waitForServerOnline(); |
| cluster.stopRegionServer(orgRegionServer.getServerName()); |
| cluster.waitForRegionServerToStop(orgRegionServer.getServerName(), 1000); |
| |
| res = executeString("select * from hbase_mapped_table1"); |
| assertResultSet(res); |
| res.close(); |
| } finally { |
| executeString("DROP TABLE hbase_mapped_table1 PURGE").close(); |
| hAdmin.close(); |
| if (htable == null) { |
| htable.close(); |
| } |
| } |
| } |
| |
| private String resultSetToString(ResultScanner scanner, byte[][] cfNames, byte[][] qualifiers, |
| boolean [] binaries, Schema schema) throws Exception { |
| StringBuilder sb = new StringBuilder(); |
| Result result = null; |
| while ( (result = scanner.next()) != null ) { |
| if (binaries[0]) { |
| sb.append(HBaseBinarySerializerDeserializer.deserialize(schema.getColumn(0), result.getRow()).asChar()); |
| } else { |
| sb.append(new String(result.getRow())); |
| } |
| |
| for (int i = 0; i < cfNames.length; i++) { |
| if (cfNames[i] == null) { |
| //rowkey |
| continue; |
| } |
| if (qualifiers[i] == null) { |
| Map<byte[], byte[]> values = result.getFamilyMap(cfNames[i]); |
| if (values == null) { |
| sb.append(", null"); |
| } else { |
| sb.append(", {"); |
| String delim = ""; |
| for (Map.Entry<byte[], byte[]> valueEntry: values.entrySet()) { |
| byte[] keyBytes = valueEntry.getKey(); |
| byte[] valueBytes = valueEntry.getValue(); |
| |
| if (binaries[i]) { |
| sb.append(delim).append("\"").append(keyBytes == null ? "" : Bytes.toLong(keyBytes)).append("\""); |
| sb.append(": \"").append(HBaseBinarySerializerDeserializer.deserialize(schema.getColumn(i), valueBytes)).append("\""); |
| } else { |
| sb.append(delim).append("\"").append(keyBytes == null ? "" : new String(keyBytes)).append("\""); |
| sb.append(": \"").append(HBaseTextSerializerDeserializer.deserialize(schema.getColumn(i), valueBytes)).append("\""); |
| } |
| delim = ", "; |
| } |
| sb.append("}"); |
| } |
| } else { |
| byte[] value = result.getValue(cfNames[i], qualifiers[i]); |
| if (value == null) { |
| sb.append(", null"); |
| } else { |
| if (binaries[i]) { |
| sb.append(", ").append(HBaseBinarySerializerDeserializer.deserialize(schema.getColumn(i), value)); |
| } else { |
| sb.append(", ").append(HBaseTextSerializerDeserializer.deserialize(schema.getColumn(i), value)); |
| } |
| } |
| } |
| } |
| sb.append("\n"); |
| } |
| |
| return sb.toString(); |
| } |
| } |
