@Test public void testGetFromDescendingOrder() throws Exception { Tuple firstTuple = null; Tuple lastTuple; TableMeta meta = CatalogUtil.newTableMeta(StoreType.CSV); Path tablePath = StorageUtil.concatPath(testDir, "testGetFromDescendingOrder", "table.csv"); fs.mkdirs(tablePath.getParent()); Appender appender = sm.getAppender(meta, schema, tablePath); appender.init(); Tuple tuple = new VTuple(schema.size()); for (int i = (TEST_TUPLE - 1); i >= 0 ; i--) { tuple.put( new Datum[] { DatumFactory.createInt4(i), DatumFactory.createInt4(i + 5) }); appender.addTuple(tuple); if (firstTuple == null) { firstTuple = new VTuple(tuple); } } lastTuple = new VTuple(tuple); appender.flush(); appender.close(); TableDesc employee = new TableDesc("employee", schema, meta, tablePath); catalog.addTable(employee); FileFragment[] frags = sm.splitNG(conf, "employee", meta, tablePath, Integer.MAX_VALUE); TaskAttemptContext ctx = new TaskAttemptContext(conf, LocalTajoTestingUtility.newQueryUnitAttemptId(), new FileFragment[] {frags[0]}, testDir); ctx.setEnforcer(new Enforcer()); Expr expr = analyzer.parse(SORT_QUERY[1]); LogicalPlan plan = planner.createPlan(expr); LogicalNode rootNode = optimizer.optimize(plan); PhysicalPlanner phyPlanner = new PhysicalPlannerImpl(conf,sm); PhysicalExec exec = phyPlanner.createPlan(ctx, rootNode); ExternalSortExec sort = null; if (exec instanceof ProjectionExec) { ProjectionExec projExec = (ProjectionExec) exec; sort = projExec.getChild(); } else if (exec instanceof ExternalSortExec) { sort = (ExternalSortExec) exec; } else { assertTrue(false); } SortSpec[] sortSpecs = sort.getPlan().getSortKeys(); RangeShuffleFileWriteExec idxStoreExec = new RangeShuffleFileWriteExec(ctx, sm, sort, sort.getSchema(), sort.getSchema(), sortSpecs); exec = idxStoreExec; exec.init(); exec.next(); exec.close(); Schema keySchema = PlannerUtil.sortSpecsToSchema(sortSpecs); TupleComparator comp = new TupleComparator(keySchema, sortSpecs); BSTIndex bst = new BSTIndex(conf); BSTIndex.BSTIndexReader reader = bst.getIndexReader( new Path(testDir, "output/index"), keySchema, comp); reader.open(); TableMeta outputMeta = CatalogUtil.newTableMeta(StoreType.RAW, new Options()); SeekableScanner scanner = StorageManagerFactory.getSeekableScanner(conf, outputMeta, schema, StorageUtil.concatPath(testDir, "output", "output")); scanner.init(); int cnt = 0; while(scanner.next() != null) { cnt++; } scanner.reset(); assertEquals(TEST_TUPLE ,cnt); Tuple keytuple = new VTuple(2); for(int i = (TEST_TUPLE - 1) ; i >= 0; i --) { keytuple.put(0, DatumFactory.createInt4(i)); keytuple.put(1, DatumFactory.createInt4(i + 5)); long offsets = reader.find(keytuple); scanner.seek(offsets); tuple = scanner.next(); assertTrue("[seek check " + (i) + " ]" , i == tuple.get(0).asInt4()); } TupleRange totalRange = new TupleRange(sortSpecs, firstTuple, lastTuple); UniformRangePartition partitioner = new UniformRangePartition(totalRange, sortSpecs, true); TupleRange [] partitions = partitioner.partition(25); File dataFile = new File((new Path(testDir, "output")).toUri()); RangeRetrieverHandler handler = new RangeRetrieverHandler( dataFile, keySchema, comp); List<Long []> offsets = new ArrayList<Long []>(); for (int i = 0; i < partitions.length; i++) { FileChunk chunk = getFileChunk(handler, keySchema, partitions[i], i == 0); offsets.add(new Long[] {chunk.startOffset(), chunk.length()}); } scanner.close(); long fileLength = new File(dataFile, "data/data").length(); Long[] previous = null; for (Long [] offset : offsets) { if (previous == null) { assertTrue(offset[0] + offset[1] == fileLength); previous = offset; continue; } assertTrue(offset[0] + offset[1] == previous[0]); previous = offset; } }
@Override public FileChunk get(Map<String, List<String>> kvs) throws IOException { File data = new File(this.file, "data/data"); byte [] startBytes = Base64.decodeBase64(kvs.get("start").get(0)); Tuple start = RowStoreUtil.RowStoreDecoder.toTuple(schema, startBytes); byte [] endBytes; Tuple end; endBytes = Base64.decodeBase64(kvs.get("end").get(0)); end = RowStoreUtil.RowStoreDecoder.toTuple(schema, endBytes); boolean last = kvs.containsKey("final"); if(!comp.isAscendingFirstKey()) { Tuple tmpKey = start; start = end; end = tmpKey; } LOG.info("GET Request for " + data.getAbsolutePath() + " (start="+start+", end="+ end + (last ? ", last=true" : "") + ")"); if (idxReader.getFirstKey() == null && idxReader.getLastKey() == null) { LOG.info("There is no contents"); return null; } if (comp.compare(end, idxReader.getFirstKey()) < 0 || comp.compare(idxReader.getLastKey(), start) < 0) { LOG.info("Out of Scope (indexed data [" + idxReader.getFirstKey() + ", " + idxReader.getLastKey() + "], but request start:" + start + ", end: " + end); return null; } long startOffset; long endOffset; try { startOffset = idxReader.find(start); } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } try { endOffset = idxReader.find(end); if (endOffset == -1) { endOffset = idxReader.find(end, true); } } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } if (startOffset == -1) { try { startOffset = idxReader.find(start, true); } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } } if (startOffset == -1) { throw new IllegalStateException("startOffset " + startOffset + " is negative \n" + "State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); } if (last || (endOffset == -1 && comp.compare(idxReader.getLastKey(), end) < 0)) { endOffset = data.length(); } FileChunk chunk = new FileChunk(data, startOffset, endOffset - startOffset); LOG.info("Retrieve File Chunk: " + chunk); return chunk; }
RangeRetrieverHandler implements RetrieverHandler { @Override public FileChunk get(Map<String, List<String>> kvs) throws IOException { File data = new File(this.file, "data/data"); byte [] startBytes = Base64.decodeBase64(kvs.get("start").get(0)); Tuple start = RowStoreUtil.RowStoreDecoder.toTuple(schema, startBytes); byte [] endBytes; Tuple end; endBytes = Base64.decodeBase64(kvs.get("end").get(0)); end = RowStoreUtil.RowStoreDecoder.toTuple(schema, endBytes); boolean last = kvs.containsKey("final"); if(!comp.isAscendingFirstKey()) { Tuple tmpKey = start; start = end; end = tmpKey; } LOG.info("GET Request for " + data.getAbsolutePath() + " (start="+start+", end="+ end + (last ? ", last=true" : "") + ")"); if (idxReader.getFirstKey() == null && idxReader.getLastKey() == null) { LOG.info("There is no contents"); return null; } if (comp.compare(end, idxReader.getFirstKey()) < 0 || comp.compare(idxReader.getLastKey(), start) < 0) { LOG.info("Out of Scope (indexed data [" + idxReader.getFirstKey() + ", " + idxReader.getLastKey() + "], but request start:" + start + ", end: " + end); return null; } long startOffset; long endOffset; try { startOffset = idxReader.find(start); } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } try { endOffset = idxReader.find(end); if (endOffset == -1) { endOffset = idxReader.find(end, true); } } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } if (startOffset == -1) { try { startOffset = idxReader.find(start, true); } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } } if (startOffset == -1) { throw new IllegalStateException("startOffset " + startOffset + " is negative \n" + "State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); } if (last || (endOffset == -1 && comp.compare(idxReader.getLastKey(), end) < 0)) { endOffset = data.length(); } FileChunk chunk = new FileChunk(data, startOffset, endOffset - startOffset); LOG.info("Retrieve File Chunk: " + chunk); return chunk; } }
RangeRetrieverHandler implements RetrieverHandler { @Override public FileChunk get(Map<String, List<String>> kvs) throws IOException { File data = new File(this.file, "data/data"); byte [] startBytes = Base64.decodeBase64(kvs.get("start").get(0)); Tuple start = RowStoreUtil.RowStoreDecoder.toTuple(schema, startBytes); byte [] endBytes; Tuple end; endBytes = Base64.decodeBase64(kvs.get("end").get(0)); end = RowStoreUtil.RowStoreDecoder.toTuple(schema, endBytes); boolean last = kvs.containsKey("final"); if(!comp.isAscendingFirstKey()) { Tuple tmpKey = start; start = end; end = tmpKey; } LOG.info("GET Request for " + data.getAbsolutePath() + " (start="+start+", end="+ end + (last ? ", last=true" : "") + ")"); if (idxReader.getFirstKey() == null && idxReader.getLastKey() == null) { LOG.info("There is no contents"); return null; } if (comp.compare(end, idxReader.getFirstKey()) < 0 || comp.compare(idxReader.getLastKey(), start) < 0) { LOG.info("Out of Scope (indexed data [" + idxReader.getFirstKey() + ", " + idxReader.getLastKey() + "], but request start:" + start + ", end: " + end); return null; } long startOffset; long endOffset; try { startOffset = idxReader.find(start); } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } try { endOffset = idxReader.find(end); if (endOffset == -1) { endOffset = idxReader.find(end, true); } } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } if (startOffset == -1) { try { startOffset = idxReader.find(start, true); } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } } if (startOffset == -1) { throw new IllegalStateException("startOffset " + startOffset + " is negative \n" + "State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); } if (last || (endOffset == -1 && comp.compare(idxReader.getLastKey(), end) < 0)) { endOffset = data.length(); } FileChunk chunk = new FileChunk(data, startOffset, endOffset - startOffset); LOG.info("Retrieve File Chunk: " + chunk); return chunk; } RangeRetrieverHandler(File outDir, Schema schema, TupleComparator comp); }
RangeRetrieverHandler implements RetrieverHandler { @Override public FileChunk get(Map<String, List<String>> kvs) throws IOException { File data = new File(this.file, "data/data"); byte [] startBytes = Base64.decodeBase64(kvs.get("start").get(0)); Tuple start = RowStoreUtil.RowStoreDecoder.toTuple(schema, startBytes); byte [] endBytes; Tuple end; endBytes = Base64.decodeBase64(kvs.get("end").get(0)); end = RowStoreUtil.RowStoreDecoder.toTuple(schema, endBytes); boolean last = kvs.containsKey("final"); if(!comp.isAscendingFirstKey()) { Tuple tmpKey = start; start = end; end = tmpKey; } LOG.info("GET Request for " + data.getAbsolutePath() + " (start="+start+", end="+ end + (last ? ", last=true" : "") + ")"); if (idxReader.getFirstKey() == null && idxReader.getLastKey() == null) { LOG.info("There is no contents"); return null; } if (comp.compare(end, idxReader.getFirstKey()) < 0 || comp.compare(idxReader.getLastKey(), start) < 0) { LOG.info("Out of Scope (indexed data [" + idxReader.getFirstKey() + ", " + idxReader.getLastKey() + "], but request start:" + start + ", end: " + end); return null; } long startOffset; long endOffset; try { startOffset = idxReader.find(start); } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } try { endOffset = idxReader.find(end); if (endOffset == -1) { endOffset = idxReader.find(end, true); } } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } if (startOffset == -1) { try { startOffset = idxReader.find(start, true); } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } } if (startOffset == -1) { throw new IllegalStateException("startOffset " + startOffset + " is negative \n" + "State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); } if (last || (endOffset == -1 && comp.compare(idxReader.getLastKey(), end) < 0)) { endOffset = data.length(); } FileChunk chunk = new FileChunk(data, startOffset, endOffset - startOffset); LOG.info("Retrieve File Chunk: " + chunk); return chunk; } RangeRetrieverHandler(File outDir, Schema schema, TupleComparator comp); @Override FileChunk get(Map<String, List<String>> kvs); }
RangeRetrieverHandler implements RetrieverHandler { @Override public FileChunk get(Map<String, List<String>> kvs) throws IOException { File data = new File(this.file, "data/data"); byte [] startBytes = Base64.decodeBase64(kvs.get("start").get(0)); Tuple start = RowStoreUtil.RowStoreDecoder.toTuple(schema, startBytes); byte [] endBytes; Tuple end; endBytes = Base64.decodeBase64(kvs.get("end").get(0)); end = RowStoreUtil.RowStoreDecoder.toTuple(schema, endBytes); boolean last = kvs.containsKey("final"); if(!comp.isAscendingFirstKey()) { Tuple tmpKey = start; start = end; end = tmpKey; } LOG.info("GET Request for " + data.getAbsolutePath() + " (start="+start+", end="+ end + (last ? ", last=true" : "") + ")"); if (idxReader.getFirstKey() == null && idxReader.getLastKey() == null) { LOG.info("There is no contents"); return null; } if (comp.compare(end, idxReader.getFirstKey()) < 0 || comp.compare(idxReader.getLastKey(), start) < 0) { LOG.info("Out of Scope (indexed data [" + idxReader.getFirstKey() + ", " + idxReader.getLastKey() + "], but request start:" + start + ", end: " + end); return null; } long startOffset; long endOffset; try { startOffset = idxReader.find(start); } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } try { endOffset = idxReader.find(end); if (endOffset == -1) { endOffset = idxReader.find(end, true); } } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } if (startOffset == -1) { try { startOffset = idxReader.find(start, true); } catch (IOException ioe) { LOG.error("State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); throw ioe; } } if (startOffset == -1) { throw new IllegalStateException("startOffset " + startOffset + " is negative \n" + "State Dump (the requested range: " + "[" + start + ", " + end+")" + ", idx min: " + idxReader.getFirstKey() + ", idx max: " + idxReader.getLastKey()); } if (last || (endOffset == -1 && comp.compare(idxReader.getLastKey(), end) < 0)) { endOffset = data.length(); } FileChunk chunk = new FileChunk(data, startOffset, endOffset - startOffset); LOG.info("Retrieve File Chunk: " + chunk); return chunk; } RangeRetrieverHandler(File outDir, Schema schema, TupleComparator comp); @Override FileChunk get(Map<String, List<String>> kvs);  }