Ignite Transaction Kullanımı

Giriş

Bir tablo 3 tane atomicity değerinden birisine sahip olabilir.

1. ATOMIC
2. TRANSACTIONAL
3. TRANSACTIONAL_SNAPSHOT

ATOMIC

Açıklaması şöyle. Yani distributed transaction kullanmaz

Specifies atomic-only cache behaviour. In this mode distributed transactions and distributed locking are not supported. Disabling transactions and locking allows to achieve much higher performance and throughput ratios.

In addition to transactions and locking, one of the main differences in ATOMIC mode is that bulk writes, such as putAll(...), removeAll(...), and transformAll(...) methods, become simple batch operations which can partially fail. In case of partial failure CachePartialUpdateException will be thrown which will contain a list of keys for which the update failed. It is recommended that bulk writes are used whenever multiple keys need to be inserted or updated in cache, as they reduce number of network trips and provide better performance.

TRANSACTIONAL

Açıklaması şöyle. Burada anlamadığım şey sadece Key-Value API (yani Thick Client) için kullanılır deniyor. JDBC'nin Key-value API'ye dahil olup olmadığı yazmıyor. Ancak neticede distributed transaction kullanıyor

Enables fully ACID-compliant transactional cache behavior for the key-value API.

Note! In this mode, transactional consistency is guaranteed for key-value API operations only. To enable ACID capabilities for SQL transactions, use the TRANSACTIONAL_SNAPSHOT mode.

Note! This atomicity mode is not compatible with the other modes within the same transaction. if a transaction is executed over multiple caches, all caches must have the same atomicity mode, either TRANSACTIONAL_SNAPSHOT or TRANSACTIONAL.

Key-Value API şöyle

try (Ignite ignite = Ignition.start("config/ignite-config.xml")) {
  IgniteCache<CityKey, City> cityCache = ignite.cache("City");

  CityKey key = new CityKey(5, "NLD");

  //getting the city by ID and country code
  City city = cityCache.get(key);

  System.out.println(">> Updating Amsterdam record:");

  city.setPopulation(city.getPopulation() - 10_000);

  cityCache.put(key, city);

  System.out.println(cityCache.get(key));
}

Key-Value API ile 2 farklı transaction kullanılabilir.

1. Pessimistic

2. Optimistic

Transaction Arayüzü yazısına bakınız

3. TRANSACTIONAL_SNAPSHOT

Açıklaması şöyle. Öncelikle bu kullanım henüz beta durumunda. Yani problem çıkartabilir. Tablodaki kayıtı MVCC haline getirir. Yani aynı Hibernate'teki gibi satıra bir sürüm numarası verir. Böylece Optimistic lock kullanılır

This is an experimental feature. Transactional SQL is currently in a beta status.

Specifies fully ACID-compliant transactional cache behavior for both key-value API and SQL transactions.

This atomicity mode enables multiversion concurrency control (MVCC) for the cache. In MVCC-enabled caches, when a transaction updates a row, it creates a new version of that row instead of overwriting it. Other users continue to see the old version of the row until the transaction is committed. In this way, readers and writers do not conflict with each other and always work with a consistent dataset. The old version of data is cleaned up when it's no longer accessed by anyone.

With this mode enabled, one node is elected as an MVCC coordinator. This node tracks all in-flight transactions and queries executed in the cluster. Each transaction or query executed over the cache with TRANSACTIONAL_SNAPSHOT mode works with a current snapshot of data generated for this transaction or query by the coordinator. This snapshot ensures that the transaction works with a consistent database state during its execution period.

Note! This atomicity mode is not compatible with the other modes within the same transaction. If a transaction is executed over multiple caches, all caches must have the same atomicity mode, either TRANSACTIONAL_SNAPSHOT or TRANSACTIONAL.

Eğer transaction içindeki bir tablo TRANSACTIONAL_SNAPSHOT geri kalan tüm tablolar da böyle olmalı Açıklaması şöyle

The TRANSACTIONAL_SNAPSHOT mode is enabled per cache and does not permit caches with different atomicity modes within one transaction. Thus, if you want to cover multiple tables in one SQL transaction, all tables must be created with the TRANSACTIONAL_SNAPSHOT mode.

FieldsQueryCursor Arayüzü - Thick Client

Giriş

Şu satırı dahil ederiz. QueryCursor arayüzünü gerçekleştirir

import org.apache.ignite.cache.query.FieldsQueryCursor;

getAll metodu

Örnek - Select

Şöyle yaparız

Ignite ignite = ...;
IgniteCache<PersonKey, Person> cache = ...;

List<List<?>> result = cache.query(
  new SqlFieldsQuery("SELECT * FROM PERSON WHERE PERSONID = 9000")).getAll();

Örnek - Select

Şöyle yaparız

String sql   = "SELECT COUNT(*) AS NUM FROM " + cacheName + " " + sqlParam;
SqlFieldsQuery query = new SqlFieldsQuery(sql);
query.setArgs(args);

FieldsQueryCursor<List<?>> cursor = cache.query(query);
resultCount = cursor.getAll().get(0).get(0);

Örnek - Select

Şöyle yaparız

try (QueryCursor<List<?>> cur = cache2.query(new SqlFieldsQuery("select _key from table"))){
  for (List<?> r : cur) {
    Long key = (Long)r.get(0);
  }
}

IgniteSet Arayüzü - Thick Client In-Memory Data Grid

Giriş

Şu satırı dahil ederiz

import org.apache.ignite.IgniteSet;

Kalıtımı şöyledir

IgniteQueue<T> extends Set<T>

add metodu

Set'i yaratmak için name + CollectionConfiguration belirtilir

Örnek

Şöyle yaparız

IgniteSet<String> set = Ignition.ignite().set("setName", null);
for (int i = 0; i < 5; i++) {
  String item = UUID.randomUUID() + "_" + i;
  set.add(item);
}

for (String item : set)
  println("Set item: " + item);

println(set.contains("1"));
println("Set size: " + set.size());
println("Removed: " + set.remove("0"));

IgniteQueue Arayüzü - Thick Client In-Memory Data Grid

Giriş

Şu satırı dahil ederiz

import org.apache.ignite.IgniteQueue;

Kalıtımı şöyledir

IgniteQueue<T> extends BlockingQueue<T>

take metodu - Blocking

Kuyruğu yaratmak için queue name + capacity + CollectionConfiguration belirtilir

Örnek

Şöyle yaparız

IgniteQueue<String> queue = Ignition.ignite().queue("queueName", 0, null);
int TIMES = 10;
for (int i = 0; i < RETRIES; i++) {
  String item = UUID.randomUUID() + "_" + i;
  queue.put(item);
  println("Queue item has been added: " + item);
}

// IgniteQueue is fully compatible with Java library.
for (String item : queue)
  println("Queue item: " + item);

// Take items from queue head.
for (int i = 0; i < TIMES; i++)
  println("Queue item has been read from queue head: " + queue.take());

// Take items from queue head once again.
for (int i = 0; i < TIMES; i++)
  println("Queue item has been read from queue head: " + queue.poll());

CREATE TABLE WITH CACHE_NAME

Giriş

Açıklaması şöyle

When the CREATE TABLE command is executed, the name of the cache is generated with the following format- SQL_{SCHEMA_NAME}_{TABLE}. Use the CACHE_NAME parameter to override the default name.

Örnek

Şöyle yaparız. Burada CACHE_NAME olarak TradeStatsCache belirtiliyor

CREATE TABLE IF NOT EXISTS PRODSCHEMA.TRADESTATS 
(
    TRADERECORDCREATIONDATE VARCHAR(100),
    PRODUCTTYPE VARCHAR(100),
    PRODUCTSUBTYPE VARCHAR(100),
    TRADETYPE VARCHAR(100),
    TOTALIMPACTEDTRADES BIGINT,
    TOTALTRADEOPENEXCEPTION BIGINT,
    TOTALTRADECOUNT BIGINT,
    UPDATEDAT TIMESTAMP,
    PRIMARY KEY (TRADERECORDCREATIONDATE,PRODUCTTYPE,PRODUCTSUBTYPE,TRADETYPE)
) WITH "template=outputDataCacheTemplate,
    CACHE_NAME=TradeStatsCache,
    AFFINITY_KEY=TRADERECORDCREATIONDATE,
    CACHE_GROUP=TRADE_METADATA_AGGREGATES,
    KEY_TYPE=com.abc.entities.trade.TradeStatsEntityKey,
    VALUE_TYPE=com.abc.entities.trade.TradeStatsEntity";

XML DataStorageConfiguration Bean

Giriş

Bu bean IgniteConfiguration bean'inin dataStorageProperty alanına denk gelir.

defaultDataRegionConfiguration Alanı

Örnek

Şöyle yaparız

<dataStorageConfiguration>
  <defaultDataRegionConfiguration
    name="Default_Region"
    initialSize="104857600"/> <!-- 100*1024*1024 -->

  <dataRegionConfigurations>
    <dataRegionConfiguration
      name="My_Region"
      initialSize="104857600"
      persistenceEnabled="true" 
      metricsEnabled="true" />
    </dataRegionConfigurations>
</dataStorageConfiguration>

defaultDataRegionConfiguration Alanı

Ne kadar bellek kullanacağını ve persistence kullanılıp kullanılmayacağını belirtiriz.

Örnek

Şöyle yaparız

<!-- Memory configuration. -->
<property name="dataStorageConfiguration">
  <bean class="org.apache.ignite.configuration.DataStorageConfiguration">
    <property name="defaultDataRegionConfiguration">
      <bean class="org.apache.ignite.configuration.DataRegionConfiguration">
        <!-- max RAM size -->
	<property name="maxSize" value="#{150 * 1024 * 1024}"/>

	<!-- memory metrics -->
	<property name="metricsEnabled" value="true"/>

	<!-- Enabling persistence. -->
	<property name="persistenceEnabled" value="true"/>
      </bean>
    </property>

    <!-- persistence metrics -->
    <property name="metricsEnabled" value="true"/>
  </bean>
</property>

wal Ayarları

• walMode 
• walSegmentSize
• writeThrottlingEnabled
• walPath
• walArchivePath

gibi bir sürü ayar yapılabilir.

walMode Alanı

FSYNC değeri alabilir

Örnek

Şöyle yaparız

<bean class="org.apache.ignite.configuration.DataStorageConfiguration">
  <property name="metricsEnabled" value="true"/>
  <property name="walMode" value="FSYNC"/>
  <property name="defaultDataRegionConfiguration">
    <bean class="org.apache.ignite.configuration.DataRegionConfiguration">
      <property name="persistenceEnabled" value="true"/>
      <property name="metricsEnabled" value="true"/>
    </bean>
  </property>

  <property name="walPath" value="/gridgain-dev/wal"/>
  <property name="walArchivePath" value="/gridgain-dev/walarchive"/>
</bean>

walSegmentSize Alanı

Örnek

Şöyle yaparız

<property name="dataStorageConfiguration">
  <bean class="org.apache.ignite.configuration.DataStorageConfiguration">
    <!-- set the size of wal segments to 128MB -->
    <property name="walSegmentSize" value="#{128 * 1024 * 1024}"/>
    <property name="writeThrottlingEnabled" value="true"/>
    <!-- Set the page size to 8 KB -->
    <property name="pageSize" value="#{8 * 1024}"/>
    <property name="defaultDataRegionConfiguration">
      <bean class="org.apache.ignite.configuration.DataRegionConfiguration">
        <property name="name" value="Default_Region"/>
	<!-- Memory region of 20 MB initial size. -->
	<property name="initialSize" value="#{20 * 1024 * 1024}"/>
	<!-- Memory region of 8 GB max size. -->
	<property name="maxSize" value="#{8L * 1024 * 1024 * 1024}"/>
	<!-- Enabling eviction for this memory region. -->
	<property name="pageEvictionMode" value="RANDOM_2_LRU"/>
	<property name="persistenceEnabled" value="true"/>
	<!-- Increasing the buffer size to 1 GB. -->
	<property name="checkpointPageBufferSize" value="#{1024L * 1024 * 1024}"/>
      </bean>
   </property>
   <property name="walPath" value="/gridgain/wal"/>
   <property name="walArchivePath" value="/gridgain/walarchive"/>
  </bean>
</property>

JDBC Connection String

Giriş

jdbc ile bağlanmak istersek şöyle yaparız

jdbc:ignite:thin://localhost:10800

JdbcThinConnection nesnesi yaratır. Bu sınıf şöyle yani TCP ile bağlantı kuruyor

public class JdbcThinConnection implements Connection {
  ...
  private volatile JdbcThinTcpIo singleIo;
  ...
}

Schema

Başka scheme' ya bağlanmak istersek şöyle yaparız

jdbc:ignite:thin://localhost:10800/MY_SCHEMA

Hangi schema' ya bağlandıysak CREATE TABLE cümlesi o schema altında tablo yaratır.

Schema ile ilgili açıklama şöyle

Here are a few options to connect to a specific schema in Ignite:

  jdbc:ignite:thin://172.16.0.5/mySchema - will connect to "MYSCHEMA" (case insensitive)

  jdbc:ignite:thin://172.16.0.5/"mySchema" - will connect to "mySchema" (case sensitive)

Transaction

JDBC ile transaction yapılabilir. Konuyla ilgili bir soru burada.

Transaction Isolation Level

Grig Gain 8.8.9 ile denediğimde şu sonucu aldım

Connection conn = ...;
DatabaseMetaData dbmd = conn.getMetaData();
int defaultTransactionIsolation = dbmd.getDefaultTransactionIsolation();
//4 yani Connection.TRANSACTION_REPEATABLE_READ

Şu kod işe yaramaz

conn.setTransactionIsolation(Connection.TRANSACTION_SERIALIZABLE);

Desteklenen Isolation Level'ları şöyle deneyebiliriz

DatabaseMetaData dbmd = conn.getMetaData();
if (dbmd.supportsTransactionIsolationLevel(Connection.TRANSACTION_REPEATABLE_READ)) {
  System.out.println("Supports TRANSACTION_REPEATABLE_READ");
}
if (dbmd.supportsTransactionIsolationLevel(Connection.TRANSACTION_SERIALIZABLE)) {
  System.out.println("Supports TRANSACTION_SERIALIZABLE");
}

ResultSet Tipleri

Grig Gain 8.8.9 ile denediğimde şu kod exception fırlattı.

Connection connection = ...;
PreparedStatement statement = connection.prepareStatement("...",
							  ResultSet.TYPE_SCROLL_SENSITIVE, 
							  ResultSet.CONCUR_UPDATABLE);

Exception şöyle

Invalid result set type (only forward is supported).

INSERT Cümlesi

Elimizde şöyle bir SQL olsun.

CREATE TABLE IF NOT EXISTS MYUSER (
  ID NUMBER(10,0),
  NAME VARCHAR2(100),
  PRIMARY KEY (ID)
) WITH "ATOMICITY=TRANSACTIONAL";

Ama aslında dokümantasyon şöyle. Aslında SQL Transaction ile TRANSACTIONAL_SNAPSHOT kullan diyor ancak bu cümle sanırım JDBC'yi kapsamıyor.

Enables fully ACID-compliant transactional cache behavior for the key-value API.

Note! In this mode, transactional consistency is guaranteed for key-value API operations only. To enable ACID capabilities for SQL transactions, use the TRANSACTIONAL_SNAPSHOT mode.

Note! This atomicity mode is not compatible with the other modes within the same transaction. if a transaction is executed over multiple caches, all caches must have the same atomicity mode, either TRANSACTIONAL_SNAPSHOT or TRANSACTIONAL.

Ancak benim denemelerimde bu tabloya şöyle yaptım ve sorun olmadan çalışıyor. Burada tablonun TRANSACTIONAL olmasında bir problem yok. Bunu denememim sebebi TRANSACTIONAL_SNAPSHOT kullandığımda MVCC'nin devreye girmesi. Ben de bunu istemiyorum

Connection connection = ...;

int id = ...;
String name = ...;
String SQL_INSERT = "INSERT INTO MYUSER(ID,NAME) VALUES(?,?)";
try (PreparedStatement insertStatement = connection.prepareStatement(SQL_INSERT)) {

  insertStatement.setLong(1, id);
  insertStatement.setString(2, name);

   updateStatement.executeUpdate();
} catch (Exception exception) {
 ...
}

Aynı şeyi tabloyu şöyle yaratarak denedim yine sorun çıkmadı

CREATE TABLE IF NOT EXISTS MYUSER (
  ID NUMBER(10,0),
  NAME VARCHAR2(100),
  PRIMARY KEY (ID)
) WITH "ATOMICITY=ATOMIC";

SELECT FOR UPDATE Cümlesi

Elimizde şöyle bir SQL olsun.

CREATE TABLE IF NOT EXISTS MYSEQUENCE (
  SEQ_NAME VARCHAR2(100),
  SEQ_VALUE NUMBER(10,0),
  PRIMARY KEY (SEQ_NAME)
) WITH "ATOMICITY=TRANSACTIONAL_SNAPSHOT";

INSERT INTO MYSEQUENCE(SEQ_NAME, SEQ_VALUE) VALUES ('SEQ1', 1);

Bu tabloda TRANSACTIONAL_SNAPSHOT kullanıldığı için SELECT FOR UPDATE yapılabilir. Şöyle yaparız

String SQL_QUERY = "SELECT SEQ_NAME, SEQ_VALUE FROM MYSEQUENCE WHERE SEQ_NAME = ?
FOR UPDATE ";
try (PreparedStatement selectStatement = conn.prepareStatement(SQL_QUERY)) {
  ...
}

Eğer tabloyu TRANSACTIONAL_SNAPSHOT olarak yaratmazsak exception alırız. Exception şöyle

SELECT FOR UPDATE query requires transactional cache with MVCC enabled.

SELECT FOR UPDATE ile bir başka transaction bizim kaydımızı halen okuyabiliyor ancak güncelleyemez.