co.cask.cdap.data2.dataset2.lib.table

Class BufferingTable

java.lang.Object

co.cask.cdap.data2.dataset2.lib.table.AbstractTable

co.cask.cdap.data2.dataset2.lib.table.BufferingTable

All Implemented Interfaces:

co.cask.cdap.api.data.batch.BatchReadable<byte[],co.cask.cdap.api.dataset.table.Row>, co.cask.cdap.api.data.batch.BatchWritable<byte[],co.cask.cdap.api.dataset.table.Put>, co.cask.cdap.api.data.batch.RecordScannable<co.cask.cdap.api.data.format.StructuredRecord>, co.cask.cdap.api.data.batch.RecordWritable<co.cask.cdap.api.data.format.StructuredRecord>, co.cask.cdap.api.dataset.Dataset, co.cask.cdap.api.dataset.metrics.MeteredDataset, co.cask.cdap.api.dataset.table.Table, co.cask.tephra.TransactionAware, Closeable, AutoCloseable

Direct Known Subclasses:

HBaseTable, InMemoryTable, LevelDBTable

public abstract class BufferingTable
extends AbstractTable
implements co.cask.cdap.api.dataset.metrics.MeteredDataset

An abstract TransactionAware implementation of Table which keeps data in memory buffer until transaction commits.

Subclasses should implement methods which deal with persistent store. This implementation merges data from persistent store and in-memory buffer for read/write operations. NOTE: this implementation does not allow storing null as a value of a column NOTE: data fetched from persisted store should never have nulls in column values: this class doesn't check that and they could be exposed to user as nulls. At the same time since null value is not allowed by this implementation this could lead to un-expected results

This implementation assumes that the table has name and conflicts are resolved on row level.

NOTE: this implementation doesn't cache any data in-memory besides changes. I.e. if you do get of same data that is not in in-memory buffer twice, two times it will try to fetch it from persistent store. Given the snapshot isolation tx model, this can be improved in future implementations.

NOTE: current implementation persists changes only at the end of transaction. Beware of OOME. There should be better implementation for MapReduce case (YMMV though, for counters/aggregations this implementation looks sweet)

NOTE: Using get(byte[], byte[], byte[], int) is generally always not efficient since it always hits the persisted store even if all needed data is in-memory buffer. See more info at method javadoc

Nested Class Summary

Nested classes/interfaces inherited from class co.cask.cdap.data2.dataset2.lib.table.AbstractTable

AbstractTable.TableScanner

Field Summary

Fields

Modifier and Type	Field and Description
protected static byte[]	DELETE_MARKER

Fields inherited from class co.cask.cdap.data2.dataset2.lib.table.AbstractTable

EMPTY_ROW_MAP

Fields inherited from interface co.cask.cdap.api.dataset.table.Table

PROPERTY_COLUMN_FAMILY, PROPERTY_CONFLICT_LEVEL, PROPERTY_READLESS_INCREMENT, PROPERTY_SCHEMA, PROPERTY_SCHEMA_ROW_FIELD, PROPERTY_TTL

Constructor Summary

Constructors

Constructor and Description
BufferingTable(String name) Creates an instance of BufferingTable with row level conflict detection, without readless increments, and no schema.
BufferingTable(String name, co.cask.cdap.api.dataset.table.ConflictDetection level) Creates an instance of BufferingTable without readless increments disabled and no schema.
BufferingTable(String name, co.cask.cdap.api.dataset.table.ConflictDetection level, boolean enableReadlessIncrements, co.cask.cdap.api.data.schema.Schema schema, String rowFieldName) Creates an instance of BufferingTable.

Method Summary

Methods

Modifier and Type	Method and Description
void	close()
boolean	commitTx()
boolean	compareAndSwap(byte[] row, byte[] column, byte[] expectedValue, byte[] newValue)
void	delete(byte[] row) NOTE: Depending on the use-case, calling this method may be much less efficient than calling same method with columns as parameters because it will require a round trip to persistent store.
void	delete(byte[] row, byte[][] columns)
co.cask.cdap.api.dataset.table.Row	get(byte[] row) NOTE: Depending on the use-case, calling this method may be much less efficient than calling same method with columns as parameters because it may always require round trip to persistent store
co.cask.cdap.api.dataset.table.Row	get(byte[] row, byte[][] columns)
co.cask.cdap.api.dataset.table.Row	get(byte[] row, byte[] startColumn, byte[] stopColumn, int limit)
List<co.cask.cdap.api.dataset.table.Row>	get(List<co.cask.cdap.api.dataset.table.Get> gets)
byte[]	getNameAsTxChangePrefix() Generates a byte array to be used as the transaction change prefix.
protected abstract NavigableMap<byte[],byte[]>	getPersisted(byte[] row, byte[][] columns) Fetches column->value pairs for set of columns from persistent store.
protected abstract NavigableMap<byte[],byte[]>	getPersisted(byte[] row, byte[] startColumn, byte[] stopColumn, int limit) Fetches column->value pairs for range of columns from persistent store.
protected List<Map<byte[],byte[]>>	getPersisted(List<co.cask.cdap.api.dataset.table.Get> gets) Fetches a list of rows from persistent store.
protected static <T> NavigableMap<byte[],T>	getRange(NavigableMap<byte[],T> rowMap, byte[] startColumn, byte[] stopColumn, int limit)
List<co.cask.cdap.api.data.batch.Split>	getSplits(int numSplits, byte[] start, byte[] stop) Fallback implementation of getSplits, SplitsUtil.primitiveGetSplits(int, byte[], byte[]).
String	getTableName()
String	getTransactionAwareName()
Collection<byte[]>	getTxChanges()
protected static <T> NavigableMap<byte[],T>	head(NavigableMap<byte[],T> map, int count)
void	increment(byte[] row, byte[][] columns, long[] amounts)
co.cask.cdap.api.dataset.table.Row	incrementAndGet(byte[] row, byte[][] columns, long[] amounts)
protected abstract void	persist(NavigableMap<byte[],NavigableMap<byte[],Update>> buff) Persists in-memory buffer.
void	postTxCommit()
void	put(byte[] row, byte[][] columns, byte[][] values) NOTE: if value is null corresponded column is deleted.
boolean	rollbackTx()
co.cask.cdap.api.dataset.table.Scanner	scan(byte[] startRow, byte[] stopRow)
co.cask.cdap.api.dataset.table.Scanner	scan(co.cask.cdap.api.dataset.table.Scan scan)
protected abstract co.cask.cdap.api.dataset.table.Scanner	scanPersisted(co.cask.cdap.api.dataset.table.Scan scan) Scans range of rows from persistent store for a given Scan.
void	setMetricsCollector(MetricsCollector metricsCollector)
void	startTx(co.cask.tephra.Transaction tx)
protected abstract void	undo(NavigableMap<byte[],NavigableMap<byte[],Update>> persisted) Undos previously persisted changes.
protected static byte[]	unwrapDeleteIfNeeded(byte[] value)
protected static NavigableMap<byte[],byte[]>	unwrapDeletes(NavigableMap<byte[],byte[]> rowMap)
protected static NavigableMap<byte[],NavigableMap<byte[],byte[]>>	unwrapDeletesForRows(NavigableMap<byte[],NavigableMap<byte[],byte[]>> rows)
void	updateTx(co.cask.tephra.Transaction transaction)
protected static byte[]	wrapDeleteIfNeeded(byte[] value)

Methods inherited from class co.cask.cdap.data2.dataset2.lib.table.AbstractTable

createSplitReader, createSplitRecordScanner, delete, delete, get, get, getRecordType, getSplits, increment, increment, incrementAndGet, incrementAndGet, put, put, write, write

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

DELETE_MARKER

protected static final byte[] DELETE_MARKER

Constructor Detail

BufferingTable

public BufferingTable(String name)

Creates an instance of BufferingTable with row level conflict detection, without readless increments, and no schema.

Parameters:

name - the name of the table

BufferingTable

public BufferingTable(String name,
              co.cask.cdap.api.dataset.table.ConflictDetection level)

Creates an instance of BufferingTable without readless increments disabled and no schema.

Parameters:

name - the name of the table

level - the conflict detection level

BufferingTable

public BufferingTable(String name,
              co.cask.cdap.api.dataset.table.ConflictDetection level,
              boolean enableReadlessIncrements,
              @Nullable
              co.cask.cdap.api.data.schema.Schema schema,
              @Nullable
              String rowFieldName)

Creates an instance of BufferingTable.

Parameters:

name - the name of the table

level - the conflict detection level

enableReadlessIncrements - whether or not readless increments are enabled

schema - the schema of the table, or null if there is no schema

rowFieldName - the name of the schema field that the row key maps to, or null if there is none

Method Detail

getTableName

public String getTableName()

Returns:

name of this table

getTransactionAwareName

public String getTransactionAwareName()

Specified by:

getTransactionAwareName in interface co.cask.tephra.TransactionAware

getNameAsTxChangePrefix

public byte[] getNameAsTxChangePrefix()

Generates a byte array to be used as the transaction change prefix. Allows implementations to override it so the change prefix more closely represents the underlying storage.

Returns:

transaction change prefix

persist

protected abstract void persist(NavigableMap<byte[],NavigableMap<byte[],Update>> buff)
                         throws Exception

Persists in-memory buffer. After this method returns we assume that data can be visible to other table clients (of course other clients may choose still not to see it based on transaction isolation logic).

Parameters:

buff - in-memory buffer to persist. Map is described as row->(column->value). Map can contain null values which means that the corresponded column was deleted

Throws:

Exception

undo

protected abstract void undo(NavigableMap<byte[],NavigableMap<byte[],Update>> persisted)
                      throws Exception

Undos previously persisted changes. After this method returns we assume that data can be visible to other table clients (of course other clients may choose still not to see it based on transaction isolation logic).

Parameters:

persisted - previously persisted changes. Map is described as row->(column->value). Map can contain null values which means that the corresponded column was deleted

Throws:

Exception

getPersisted

protected abstract NavigableMap<byte[],byte[]> getPersisted(byte[] row,
                                       @Nullable
                                       byte[][] columns)
                                                     throws Exception

Fetches column->value pairs for set of columns from persistent store. NOTE: persisted store can also be in-memory, it is called "persisted" to distinguish from in-memory buffer.

Parameters:

row - row key defines the row to fetch columns from

columns - set of columns to fetch. null means fetch everything; empty array which means fetch nothing.

Returns:

map of column->value pairs, never null.

Throws:

Exception

getPersisted

protected abstract NavigableMap<byte[],byte[]> getPersisted(byte[] row,
                                       byte[] startColumn,
                                       byte[] stopColumn,
                                       int limit)
                                                     throws Exception

Fetches column->value pairs for range of columns from persistent store. NOTE: persisted store can also be in-memory, it is called "persisted" to distinguish from in-memory buffer. NOTE: Using this method is generally always not efficient since it always hits the persisted store even if all needed data is in-memory buffer. Since columns set is not strictly defined the implementation always looks up for more columns in persistent store.

Parameters:

row - row key defines the row to fetch columns from

startColumn - first column in a range, inclusive

stopColumn - last column in a range, exclusive

limit - max number of columns to fetch

Returns:

map of column->value pairs, never null.

Throws:

Exception

scanPersisted

protected abstract co.cask.cdap.api.dataset.table.Scanner scanPersisted(co.cask.cdap.api.dataset.table.Scan scan)
                                                                 throws Exception

Scans range of rows from persistent store for a given Scan. NOTE: persisted store can also be in-memory, it is called "persisted" to distinguish from in-memory buffer.

Parameters:

scan - scan configuration

Returns:

instance of Scanner, never null

Throws:

Exception

getPersisted

protected List<Map<byte[],byte[]>> getPersisted(List<co.cask.cdap.api.dataset.table.Get> gets)
                                         throws Exception

Fetches a list of rows from persistent store. Subclasses should override this if they can batch multiple gets into a single request, as the default implementation simply loops through the gets and calls getPersisted(byte[], byte[][]) on each get. NOTE: persisted store can also be in-memory, it is called "persisted" to distinguish from in-memory buffer.

Parameters:

gets - list of gets to perform

Returns:

list of rows, one for each get

Throws:

Exception

setMetricsCollector

public void setMetricsCollector(MetricsCollector metricsCollector)

Specified by:

setMetricsCollector in interface co.cask.cdap.api.dataset.metrics.MeteredDataset

close

public void close()
           throws IOException

Specified by:

close in interface Closeable

Specified by:

close in interface AutoCloseable

Throws:

IOException

startTx

public void startTx(co.cask.tephra.Transaction tx)

Specified by:

startTx in interface co.cask.tephra.TransactionAware

updateTx

public void updateTx(co.cask.tephra.Transaction transaction)

Specified by:

updateTx in interface co.cask.tephra.TransactionAware

getTxChanges

public Collection<byte[]> getTxChanges()

Specified by:

getTxChanges in interface co.cask.tephra.TransactionAware

commitTx

public boolean commitTx()
                 throws Exception

Specified by:

commitTx in interface co.cask.tephra.TransactionAware

Throws:

Exception

postTxCommit

public void postTxCommit()

Specified by:

postTxCommit in interface co.cask.tephra.TransactionAware

rollbackTx

public boolean rollbackTx()
                   throws Exception

Specified by:

rollbackTx in interface co.cask.tephra.TransactionAware

Throws:

Exception

get

public co.cask.cdap.api.dataset.table.Row get(byte[] row)

NOTE: Depending on the use-case, calling this method may be much less efficient than calling same method with columns as parameters because it may always require round trip to persistent store

Specified by:

get in interface co.cask.cdap.api.dataset.table.Table

get

public co.cask.cdap.api.dataset.table.Row get(byte[] row,
                                     byte[][] columns)

Specified by:

get in interface co.cask.cdap.api.dataset.table.Table

get

public co.cask.cdap.api.dataset.table.Row get(byte[] row,
                                     byte[] startColumn,
                                     byte[] stopColumn,
                                     int limit)

Specified by:

get in interface co.cask.cdap.api.dataset.table.Table

get

public List<co.cask.cdap.api.dataset.table.Row> get(List<co.cask.cdap.api.dataset.table.Get> gets)

Specified by:

get in interface co.cask.cdap.api.dataset.table.Table

Overrides:

get in class AbstractTable

put

public void put(byte[] row,
       byte[][] columns,
       byte[][] values)

NOTE: if value is null corresponded column is deleted. It will not be in result set when reading. Also see Table.put(byte[], byte[][], byte[][]).

Specified by:

put in interface co.cask.cdap.api.dataset.table.Table

delete

public void delete(byte[] row)

NOTE: Depending on the use-case, calling this method may be much less efficient than calling same method with columns as parameters because it will require a round trip to persistent store.

Specified by:

delete in interface co.cask.cdap.api.dataset.table.Table

delete

public void delete(byte[] row,
          byte[][] columns)

Specified by:

delete in interface co.cask.cdap.api.dataset.table.Table

incrementAndGet

public co.cask.cdap.api.dataset.table.Row incrementAndGet(byte[] row,
                                                 byte[][] columns,
                                                 long[] amounts)

Specified by:

incrementAndGet in interface co.cask.cdap.api.dataset.table.Table

increment

public void increment(byte[] row,
             byte[][] columns,
             long[] amounts)

Specified by:

increment in interface co.cask.cdap.api.dataset.table.Table

compareAndSwap

public boolean compareAndSwap(byte[] row,
                     byte[] column,
                     byte[] expectedValue,
                     byte[] newValue)

Specified by:

compareAndSwap in interface co.cask.cdap.api.dataset.table.Table

getSplits

public List<co.cask.cdap.api.data.batch.Split> getSplits(int numSplits,
                                                byte[] start,
                                                byte[] stop)

Fallback implementation of getSplits, SplitsUtil.primitiveGetSplits(int, byte[], byte[]). Ideally should be overridden by subclasses.

Specified by:

getSplits in interface co.cask.cdap.api.dataset.table.Table

Parameters:

numSplits - Desired number of splits. If greater than zero, at most this many splits will be returned. If less or equal to zero, any number of splits can be returned.

start - If non-null, the returned splits will only cover keys that are greater or equal.

stop - If non-null, the returned splits will only cover keys that are less.

Returns:

list of Split

scan

public co.cask.cdap.api.dataset.table.Scanner scan(byte[] startRow,
                                          byte[] stopRow)

Specified by:

scan in interface co.cask.cdap.api.dataset.table.Table

scan

public co.cask.cdap.api.dataset.table.Scanner scan(co.cask.cdap.api.dataset.table.Scan scan)

Specified by:

scan in interface co.cask.cdap.api.dataset.table.Table

getRange

protected static <T> NavigableMap<byte[],T> getRange(NavigableMap<byte[],T> rowMap,
                                  byte[] startColumn,
                                  byte[] stopColumn,
                                  int limit)

head

protected static <T> NavigableMap<byte[],T> head(NavigableMap<byte[],T> map,
                              int count)

wrapDeleteIfNeeded

protected static byte[] wrapDeleteIfNeeded(byte[] value)

unwrapDeleteIfNeeded

protected static byte[] unwrapDeleteIfNeeded(byte[] value)

unwrapDeletesForRows

protected static NavigableMap<byte[],NavigableMap<byte[],byte[]>> unwrapDeletesForRows(NavigableMap<byte[],NavigableMap<byte[],byte[]>> rows)

unwrapDeletes

protected static NavigableMap<byte[],byte[]> unwrapDeletes(NavigableMap<byte[],byte[]> rowMap)